kangshifu-board/scripts/collect-five-topic-multisource.py

"""Discover original pages across publishers, community and official guidance.

Search pages never become evidence. Keep every response byte and precise body
locator, reject challenge pages, and stop the affected host rather than bypass it.
"""
from concurrent.futures import ThreadPoolExecutor,as_completed
from datetime import datetime,timezone
import importlib.util
import json
from pathlib import Path
import re
import time
import urllib.parse
from lxml import html

ROOT=Path(__file__).resolve().parents[1]
WORK=ROOT/'work/five-topic-research-20261111'
SEARCH=WORK/'search';PUBLIC=WORK/'sources'
for d in [SEARCH,PUBLIC]:d.mkdir(exist_ok=True)
spec=importlib.util.spec_from_file_location('public_base_multi',ROOT/'scripts/collect-five-topic-public.py')
base=importlib.util.module_from_spec(spec);spec.loader.exec_module(base)
spec=importlib.util.spec_from_file_location('industry_base_multi',ROOT/'scripts/collect-five-topic-industry-supplement.py')
industry=importlib.util.module_from_spec(spec);spec.loader.exec_module(industry)
QUERIES={
 'wellness':['药食同源 方便面 配料 标准','菌菇 非油炸 方便面 测评','方便面 减盐 钠 工艺','当归 枸杞 方便面 汤料'],
 'student':['宿舍 方便面 开水 不用煮 测评','夜宵 方便面 小份 测评','学生 拌面 沥水 烫手','大学生 方便面 消费 调查'],
 'premium':['高端 方便面 真肉 冻干 工艺','拉面说 牛肉 方便 测评','方便面 高汤 杀菌 工艺','方便面 高端化 价格 复购'],
 'stockpile':['方便面 保质期 储存 油脂 氧化','方便面 囤货 组合装 测评','方便面 包装 阻氧 防潮','方便面 家庭 囤货 口味'],
 'novelty':['方便面 香菜 冬阴功 新品','拌面 联名 新口味 产品','方便面 地域风味 调味 研发','方便面 酸辣 藤椒 测评'],
 'policy':['食药物质 人参 西洋参 黄芪 公告 site:gov.cn','食药物质 管理规定 普通食品 功效 site:gov.cn',
           '预包装食品 营养标签 减少钠 问答 site:gov.cn','食品接触材料 塑料 微波 加热 方便面 site:gov.cn',
           '方便面 食品安全 消费提示 储存 site:gov.cn','预包装食品 保质期 储存条件 标签 问答 site:gov.cn',
           '食品添加剂 调味面制品 使用标准 问答 site:gov.cn','食品 营养声称 低钠 减钠 site:gov.cn'],
}
PUBLISHER_QUERIES=['学生 方便面','夜宵','宿舍','方便面 冻干','方便面 减盐','方便面 工艺','方便面 包装','方便面 新品','药食同源 方便面']
STATE={'schema':'ksf.five-topic.multisource-collection.v1','startedAt':base.now(),'status':'running',
       'searches':[],'sources':[],'rejected':[],'hostStops':{},'productionWrites':0}
def persist():STATE['updatedAt']=base.now();base.save(WORK/'public-collection-result.json',STATE)
def text(n):return re.sub(r'\s+',' ',n.text_content()).strip()

def search(job):
 index,topic,query=job;key='web-'+str(index).zfill(2)
 url='https://www.so.com/s?'+urllib.parse.urlencode({'q':query})
 raw,receipt=base.fetch(url,key,SEARCH)
 root=html.fromstring(raw.decode('utf8',errors='replace'))
 if receipt['httpStatus']!=200 or re.search('验证|验证码',root.xpath('string(//title)')):raise ValueError('SEARCH_UNAVAILABLE')
 found={}
 for a in root.xpath('//a[@data-mdurl]'):
  u=a.get('data-mdurl');title=text(a)
  if not u or len(title)<7 or not u.startswith(('http://','https://')):continue
  host=urllib.parse.urlparse(u).hostname or ''
  if topic=='policy' and not host.endswith('.gov.cn'):continue
  if not re.search('方便面|泡面|拌面|即食面|拉面|食药|食物|食品|原料|食用|药食|黄芪|西洋参',title):continue
  found[u]={'url':u,'title':title,'topic':topic,'query':query,'searchReceipt':key+'.http.json'}
 result={'topic':topic,'query':query,'httpReceipt':receipt,'results':list(found.values())}
 base.save(SEARCH/(key+'.results.json'),result)
 return result

def publisher_search(job):
 i,query=job;key='publisher-'+str(i).zfill(2)
 url='https://www.foodaily.com/search/articles?'+urllib.parse.urlencode({'search':query})
 raw,receipt=base.fetch(url,key,SEARCH);root=html.fromstring(raw.decode('utf8',errors='replace'))
 rows={}
 for a in root.xpath('//a[@href]'):
  u=a.get('href');title=text(a)
  if re.fullmatch(r'https://www\.foodaily\.com/articles/\d+',u or '') and re.search('方便面|泡面|拌面|汤面|面饼|拉面|速食面',title):
   rows[u]={'url':u,'title':title,'query':query,'searchReceipt':key+'.http.json'}
 result={'query':query,'httpReceipt':receipt,'results':list(rows.values())}
 base.save(SEARCH/(key+'.results.json'),result);return result

def canonical(url):
 p=urllib.parse.urlparse(url)
 return urllib.parse.urlunparse((p.scheme,p.netloc,p.path.rstrip('/'),'','',''))

def collect(candidate):
 u=candidate['url'];host=urllib.parse.urlparse(u).hostname or ''
 if host in STATE['hostStops']:raise ValueError('HOST_STOPPED_AFTER_CHALLENGE')
 key=base.digest(u.encode())[:24];cached=PUBLIC/(key+'.source.json')
 if cached.exists():return json.loads(cached.read_bytes())
 raw,receipt=base.fetch(u,key,PUBLIC)
 if receipt['httpStatus'] in (202,403,412,429) or re.search('captcha|geetest|/verify|risk_handler',receipt['finalUrl'],re.I):
  STATE['hostStops'][host]={'reason':'ACCESS_CHALLENGE','url':u,'httpStatus':receipt['httpStatus']}
  raise ValueError('ACCESS_CHALLENGE')
 if receipt['httpStatus']!=200:raise ValueError('HTTP_'+str(receipt['httpStatus']))
 if 'pdf' in receipt['contentType'].lower() or raw[:4]==b'%PDF':raise ValueError('PDF_REQUIRES_SEPARATE_EXTRACTION')
 try:decoded=raw.decode('utf8');encoding='utf8'
 except UnicodeDecodeError:decoded=raw.decode('gb18030');encoding='gb18030'
 root=html.fromstring(decoded);title=re.sub(r'\s+',' ',root.xpath('string(//title)')).strip()
 if re.search('安全验证|访问验证|验证码|访问异常|验证中心|页面不存在|404|登录',title):raise ValueError('VERIFICATION_OR_ERROR_PAGE')
 final_host=urllib.parse.urlparse(receipt['finalUrl']).hostname or ''
 is_gov=final_host.endswith('.gov.cn')
 if final_host.endswith('so.com') or final_host.endswith('sogou.com'):raise ValueError('SEARCH_REDIRECT_NOT_ORIGINAL')
 selectors=["//*[@id='UCAP-CONTENT']","//*[@id='zoom']","//*[@id='zoomcon']","//*[@id='content']",
 "//*[contains(concat(' ',normalize-space(@class),' '),' TRS_Editor ')]",
 "//*[contains(concat(' ',normalize-space(@class),' '),' article-content ')]",
 "//*[contains(concat(' ',normalize-space(@class),' '),' article_content ')]",
 "//*[contains(concat(' ',normalize-space(@class),' '),' pages_content ')]",
 "//*[@id='js_content']","//*[@id='artibody']","//*[contains(concat(' ',normalize-space(@class),' '),' post_text ')]",
 "//*[contains(concat(' ',normalize-space(@class),' '),' article-body ')]",'//article']
 if final_host=='www.foodaily.com':
  title,body,selector,published,encoding=industry.extract(raw,receipt)
 else:
  body=selector=None
  for s in selectors:
   nodes=root.xpath(s)
   if len(nodes)==1:
    value=base.paragraphs(nodes[0])
    if len(value)>=120:body=value;selector=s;break
  if not body:raise ValueError('UNIQUE_BODY_NOT_LOCATED')
  h1=root.xpath('//article//h1|//h1')
  if len(h1)==1:title=text(h1[0]) or title
  dates=root.xpath('//meta[@property="article:published_time" or @name="PubDate" or @name="pubdate" or @name="publishdate" or @name="publish-date" or @name="PublishDate"]/@content')
  if not dates:
   visible=re.search(r'\d{4}[-年/.]\d{1,2}[-月/.]\d{1,2}(?:日)?(?:\s+\d{2}:\d{2}(?::\d{2})?)?',root.text_content()[:5000])
   dates=[visible.group(0)] if visible else []
  published=dates[0] if dates else None
 if not re.search('方便面|泡面|即食面|速食面|拌面|拉面说|满汉|速达面馆|火鸡面',title+'\n'+body) and not is_gov:raise ValueError('NOT_NOODLE_RESEARCH')
 if len(body)>95000:raise ValueError('BODY_TOO_LARGE_REQUIRES_REVIEW')
 if not published:raise ValueError('PUBLICATION_DATE_NOT_LOCATED')
 parsed=re.search(r'(\d{4})[-年/.](\d{1,2})[-月/.](\d{1,2})',published)
 if parsed:
  day='-'.join([parsed[1],parsed[2].zfill(2),parsed[3].zfill(2)])
  if day>datetime.now().strftime('%Y-%m-%d'):raise ValueError('FUTURE_PUBLICATION_DATE')
 if final_host=='post.smzdm.com':
  m=re.fullmatch(r'/(?:talk/)?p/([a-z0-9]+)/?',urllib.parse.urlparse(receipt['finalUrl']).path)
  if not m:raise ValueError('EXACT_POST_URL_REQUIRED')
  platform,pid='smzdm',m[1]
 elif final_host=='www.foodaily.com':platform,pid='foodaily',urllib.parse.urlparse(receipt['finalUrl']).path.split('/')[-1]
 else:platform,pid=('govcn' if is_gov else final_host.replace('www.','').replace('.','_')),base.digest(canonical(receipt['finalUrl']).encode())[:32]
 row={'schema':'ksf.five-topic.public-source.v1','url':receipt['finalUrl'],'title':title,'body':body,'bodySha256':base.digest(body.encode()),
 'platform':platform,'postId':pid,'sourceId':platform+':'+pid,'sourceType':'official_policy_or_guidance' if is_gov else 'unreviewed_original_article',
 'publishedAtRaw':published,'collectedAt':receipt['collectedAt'],'httpReceipt':receipt,'sourceEncoding':encoding,
 'bodyExtractionSelector':selector,'collectionDirectory':'sources','searchDiscovery':candidate,'topics':[],'publishedToDatabase':False,
 'limits':['保留原作者、日期及原文，不等同独立消费者或电商购买评价','原文工艺、价格、营养及功效声明须区分来源和适用边界，不能自动作为已验证结论']}
 base.save(cached,row);return row

def main():
 inv=json.loads((ROOT/'work/library-cleanup-20261111/round-02/inventory-after-result.json').read_bytes())['body']['result']
 existing={canonical(r.get('url') or '') for r in inv['social']}
 candidates={};jobs=[(i,t,q) for i,(t,q) in enumerate([(t,q) for t,qs in QUERIES.items() for q in qs],1)]
 with ThreadPoolExecutor(max_workers=3) as pool:
  tasks={pool.submit(search,j):j for j in jobs}
  tasks.update({pool.submit(publisher_search,(i,q)):('publisher',q) for i,q in enumerate(PUBLISHER_QUERIES,1)})
  for future in as_completed(tasks):
   try:
    result=future.result();STATE['searches'].append({'query':result['query'],'count':len(result['results'])})
    for row in result['results']:
     if canonical(row['url']) not in existing:candidates.setdefault(row['url'],row)
   except Exception as e:STATE['rejected'].append({'job':tasks[future],'reason':str(e)})
   persist()
 STATE['candidateCount']=len(candidates);persist();base.save(WORK/'discovered-candidates.json',list(candidates.values()))
 print(json.dumps({'stage':'search-complete','candidates':len(candidates)}),flush=True)
 with ThreadPoolExecutor(max_workers=3) as pool:
  tasks={pool.submit(collect,row):row for row in candidates.values()}
  for future in as_completed(tasks):
   try:STATE['sources'].append(future.result())
   except Exception as e:STATE['rejected'].append({'url':tasks[future]['url'],'reason':str(e)})
   persist()
   if (len(STATE['sources'])+len(STATE['rejected']))%10==0:print(json.dumps({'stage':'originals','readable':len(STATE['sources']),'rejected':len(STATE['rejected'])}),flush=True)
 STATE.update(status='collected-awaiting-content-review',finishedAt=base.now(),uniqueSources=len(STATE['sources']));persist()
 print(json.dumps({'stage':'completed','readableOriginals':len(STATE['sources']),'candidates':len(candidates)}),flush=True)

if __name__=='__main__':main()