使用Python的Google趋势：收集兴趣和热门话题

python -m pip install mcp==2.2.0 requests==2.34.2
pnpm add scrapeless-mcp-server@0.6.3

import argparse
import asyncio
import json
import os
from pathlib import Path
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main(capture):
    key = os.environ['SCRAPELESS_KEY']
    server = StdioServerParameters(
        command='node',
        args=['--input-type=module', '-e',
              'console.log = () => {}; await import(process.argv[1]);',
              str(Path('node_modules/scrapeless-mcp-server/build/index.js').resolve())],
        env={**os.environ, 'SCRAPELESS_KEY': key})
    async with stdio_client(server) as streams:
        async with ClientSession(*streams) as session:
            await session.initialize()
            listed = await session.list_tools()
            tool = next(t for t in listed.tools if t.name == 'google_trends')
            Path('trends-tool-schema.json').write_text(
                json.dumps(tool.input_schema, indent=2))
            print(json.dumps({'discovered_tools': len(listed.tools),
                              'selected_tool': tool.name}))
            if capture:
                if key == 'metadata-discovery-only':
                    raise ValueError('A real key is required for remote capture')
                args = {'q': 'coffee', 'data_type': 'interest_over_time',
                        'date': 'today 12-m', 'geo': 'US',
                        'hl': 'en', 'tz': '0', 'cat': '0'}
                result = await session.call_tool(tool.name, args)
                Path('trends-result.json').write_text(json.dumps({
                    'request': args, 'result': result.model_dump(mode='json')
                }, indent=2))
                if result.is_error:
                    raise ValueError('Tool error; inspect saved result')
                print('Raw result saved; inspect its actual fields before mapping a series.')

parser = argparse.ArgumentParser()
parser.add_argument('--capture', action='store_true')
asyncio.run(main(parser.parse_args().capture))

import csv
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from xml.etree import ElementTree as ET
import requests

url = 'https://trends.google.com/trending/rss?geo=US'
response = requests.get(url, timeout=30)
response.raise_for_status()
root = ET.fromstring(response.content)
channel = root.find('channel')
if channel is None:
    raise ValueError('Expected RSS channel, received another document')
observed = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
folder = Path('trending-rss') / observed
folder.mkdir(parents=True, exist_ok=True)
(folder / 'source.xml').write_bytes(response.content)
ns = {'ht': 'https://trends.google.com/trending/rss'}
rows = []
for item in channel.findall('item')[:5]:
    title = (item.findtext('title') or '').strip()
    link = (item.findtext('link') or '').strip()
    if not title or not link:
        raise ValueError('RSS item missing title or link')
    rows.append({'title': title, 'link': link,
                 'published': item.findtext('pubDate'),
                 'traffic_bucket': item.findtext('ht:approx_traffic', namespaces=ns),
                 'geo': 'US', 'observed_at': observed})
if not rows:
    raise ValueError('No items; inspect source before treating this as valid empty data')
(folder / 'records.json').write_text(json.dumps(rows, ensure_ascii=False, indent=2))
with (folder / 'records.csv').open('w', newline='', encoding='utf-8') as out:
    writer = csv.DictWriter(out, fieldnames=list(rows[0]))
    writer.writeheader()
    writer.writerows(rows)
print(json.dumps({'source': response.url, 'items_saved': len(rows),
                  'source_sha256': hashlib.sha256(response.content).hexdigest()}))