#!/usr/bin/env python3 # -*- coding: utf-8 -*- """行业情报采集（DeepSeek Search 源） 用法：python intel_ds_search.py [YYYYMMDD] [daily|first] 输出：数据/market_intel/v01/ds_search_<日期>.csv + _summary.json 口径：有 URL + 有日期才入库；权威度 >=2；相关度 >=1.5；QPS 5 → 每次请求间隔 0.25s """ import csv import datetime as dt import json import os import re import sys import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import deepseek_search as DS # noqa: E402 W = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' OUTDIR = os.path.join(W, '数据/market_intel/v01') os.makedirs(OUTDIR, exist_ok=True) TARGET = sys.argv[1] if len(sys.argv) > 1 else dt.date.today().strftime('%Y%m%d') MODE = sys.argv[2] if len(sys.argv) > 2 else 'daily' T = dt.datetime.strptime(TARGET, '%Y%m%d').date() if MODE == 'daily': WIN_START, WIN_END = T - dt.timedelta(days=3), T else: WIN_START, WIN_END = T - dt.timedelta(days=1095), T WL = json.load(open(os.path.join(OUTDIR, 'watchlist_v2.json'), encoding='utf-8'))['items'] ENTITIES = [] for it in WL: ENTITIES.append((it['名称'], it.get('层', '行业龙头'))) for c in ['诺唯赞', '翌圣生物', '全式金', '菲鹏生物', '迈迪安', '宝生物', '赛默飞', '罗氏诊断']: ENTITIES.append((c, '竞品')) EVENT_KW = { '获证/注册': ['注册证', '获证', '批准', 'NMPA', '备案'], '中标/采购': ['中标', '成交', '招标', '采购', '遴选'], '产能/投资': ['扩产', '投产', '产能', '融资', '投资', '并购'], '产品/合作': ['新品', '发布', '上市', '合作', '签约', '战略'], } DOMAIN_KW = ['诊断', '检测', '试剂', '核酸', 'PCR', '分子', '生物', '医疗', '基因', '抗体', '酶'] TENDER_KW = ['招标', '中标', '成交', '采购', '遴选', '询价', '结果公告', '资格预审'] SKIP_HOST = ['zhihu.com', 'baidu.com', 'sohu.com', 'toutiao.com', 'douyin.com', 'bilibili.com', 'eastmoney.com', 'stockstar.com', '10jqka.com.cn', 'xueqiu.com', 'jrj.com.cn', 'data.eastmoney.com', 'emweb.securities.eastmoney.com'] def throttle(): time.sleep(0.26) def norm_date(s): if not s: return None try: return dt.datetime.fromisoformat(s.replace('Z', '+00:00')).date() except Exception: m = re.match(r'(\d{4})-(\d{2})-(\d{2})', s) return dt.date(*map(int, m.groups())) if m else None def classify(title, summary): text = title + ' ' + (summary or '') for ev, kws in EVENT_KW.items(): if any(k in text for k in kws): return ev return '其他动态' def lane(text): for k, v in [('qPCR/PCR', ['PCR', 'qPCR', '扩增']), ('测序', ['测序', 'NGS']), ('化学发光/免疫', ['化学发光', '免疫', '发光']), ('原料/酶', ['酶', '原料', '蛋白']), ('提取/样本', ['提取', '核酸释放', '采样'])]: if any(x.lower() in text.lower() for x in v): return k return '其他' rows, seen, seen_t, health = [], set(), set(), [] SCORE_MIN_ENTITY = 0.5 # 主体类查询相关度普遍偏低，1.5 会误杀真信号 SCORE_MIN_TENDER = 1.2 def add(entity, layer, r, ev_override=None, must_contain=None): url = (r.get('url') or '').strip() if not url.startswith('http'): return False if any(h in url for h in SKIP_HOST): return False d = norm_date(r.get('date')) if not d or not (WIN_START <= d <= WIN_END): return False smin = SCORE_MIN_TENDER if ev_override == '中标/采购' else SCORE_MIN_ENTITY if (r.get('rerank_score') or 0) < smin or (r.get('authority') or 0) < 2: return False if must_contain: body = (r.get('title') or '') + ' ' + (r.get('snippet') or '') + ' ' + (r.get('summary') or '') if must_contain not in body: return False nt = re.sub(r'[\s\W]+', '', (r.get('title') or ''))[:26] if nt in seen_t: return False key = (url, (r.get('title') or '')[:40]) if key in seen: return False seen.add(key); seen_t.add(nt) title = (r.get('title') or '').strip() if len(title) < 10 or re.match(r'^\d{6}\b', title): return False if any(k in title for k in ['行情', '股价', '资金流向', '概念股', '十大核心标的', '股票']): return False text = title + ' ' + (r.get('snippet') or '') ev = ev_override or classify(title, r.get('snippet')) rows.append({ '日期': d.isoformat(), '是否新增': '是', '需跟进': '否' if ev == '其他动态' else '是', '事件类型': ev, '赛道': lane(text), '品牌/主体': entity, '信源': r.get('site_name') or 'DeepSeek Search', '标题': title, '发布日期': d.isoformat(), '发布日期为推断': '否', 'URL': url, '相关度': r.get('rerank_score'), '权威度': r.get('authority'), '依据': '主体名「%s」出现在标题/摘要' % entity, }) return True for name, layer in ENTITIES: q = '"%s" 注册证 OR 中标 OR 扩产 OR 合作 after:%s before:%s' % ( name, WIN_START.isoformat(), (WIN_END + dt.timedelta(days=1)).isoformat()) try: rs = DS.search(q, count=10) kept = sum(1 for r in rs if add(name, layer, r, must_contain=name)) health.append({'实体': name, '查询': q[:60], '返回': len(rs), '保留': kept, '状态': 'OK'}) except Exception as e: health.append({'实体': name, '查询': q[:60], '返回': 0, '保留': 0, '状态': 'ERR %s' % str(e)[:60]}) throttle() # 招投标 for q in ['PCR 试剂 招标 OR 中标 after:%s' % WIN_START.isoformat(), '核酸检测 试剂 采购 after:%s' % WIN_START.isoformat(), '分子诊断 试剂 中标 after:%s' % WIN_START.isoformat(), '检验试剂 耗材 中标 after:%s' % WIN_START.isoformat()]: try: rs = DS.search(q, count=10) kept = 0 for r in rs: t = (r.get('title') or '') + ' ' + (r.get('snippet') or '') if not any(k in t for k in DOMAIN_KW): continue if not any(k in t for k in TENDER_KW): continue if add('招投标（二级来源）', '招投标', r, ev_override='中标/采购'): kept += 1 health.append({'实体': '招投标', '查询': q[:60], '返回': len(rs), '保留': kept, '状态': 'OK'}) except Exception as e: health.append({'实体': '招投标', '查询': q[:60], '返回': 0, '保留': 0, '状态': 'ERR %s' % str(e)[:60]}) throttle() csv_path = os.path.join(OUTDIR, 'ds_search_%s.csv' % TARGET) with open(csv_path, 'w', newline='', encoding='utf-8-sig') as f: cols = ['日期', '是否新增', '需跟进', '事件类型', '赛道', '品牌/主体', '信源', '标题', '发布日期', '发布日期为推断', 'URL', '相关度', '权威度', '依据'] w = csv.DictWriter(f, fieldnames=cols) w.writeheader() for r in sorted(rows, key=lambda x: (-(x['相关度'] or 0), x['品牌/主体'])): w.writerow(r) json.dump({'target': TARGET, 'mode': MODE, 'window': [WIN_START.isoformat(), WIN_END.isoformat()], 'rows': len(rows), 'health': health}, open(os.path.join(OUTDIR, 'ds_search_%s_summary.json' % TARGET), 'w'), ensure_ascii=False, indent=1) print('窗口 %s ~ %s ｜查询 %d 次 ｜入库 %d 条 → %s' % ( WIN_START, WIN_END, len(health), len(rows), os.path.basename(csv_path))) for r in rows[:8]: print(' [%s] %s | %s | %.1f | %s' % (r['事件类型'], r['品牌/主体'], r['标题'][:42], r['相关度'], r['URL'][:60]))