kangshifu-board/scripts/collect-five-topic-bulk-20261111.py

"""Bounded public-source expansion using readable publisher catalogs and originals.

Existing search verification responses are not retried or bypassed. Discovery
links never become content records; each selected page keeps exact HTTP bytes.
"""
from concurrent.futures import ThreadPoolExecutor,as_completed
from datetime import datetime
import importlib.util
import json
from pathlib import Path
import re
import threading
import time
import urllib.parse
from lxml import html

ROOT=Path(__file__).resolve().parents[1]
WORK=ROOT/'work/five-topic-bulk-20261111'
for name in ['sources','search']: (WORK/name).mkdir(parents=True,exist_ok=True)
spec=importlib.util.spec_from_file_location('bulkbase',ROOT/'scripts/collect-five-topic-public.py')
b=importlib.util.module_from_spec(spec);spec.loader.exec_module(b)
spec=importlib.util.spec_from_file_location('bulkindustry',ROOT/'scripts/collect-five-topic-industry-supplement.py')
industry=importlib.util.module_from_spec(spec);spec.loader.exec_module(industry)
state={'schema':'ksf.five-topic.bulk-discovery.v1','status':'running','startedAt':b.now(),'sources':[],'rejected':[],'searches':[],'hostStops':{},'productionWrites':0}
guard=threading.Lock();next_start={}
NOODLE=re.compile('方便面|泡面|拌面|即食面|速食面|火鸡面|汤达人|拉面说|速达面馆|满汉大餐|冻干面')

def persist():b.save(WORK/'collection-state.json',{**state,'updatedAt':b.now()})
def canonical(url):
 p=urllib.parse.urlparse(url);return urllib.parse.urlunparse((p.scheme,p.netloc,p.path.rstrip('/'),'','',''))
def original_post(url):
 p=urllib.parse.urlparse(urllib.parse.urljoin('https://post.smzdm.com/',url))
 if p.hostname!='post.smzdm.com' or not re.fullmatch(r'/(?:talk/)?p/[a-z0-9]+/?',p.path):return None
 return 'https://post.smzdm.com'+p.path.rstrip('/')+'/'
def decode(raw):
 try:return raw.decode('utf8'),'utf8'
 except UnicodeDecodeError:return raw.decode('gb18030'),'gb18030'
def challenge(raw,receipt,tree):
 return receipt['httpStatus'] in (202,403,412,429) or bool(re.search('captcha|/verify|risk_handler|geetest',receipt['finalUrl'],re.I)) or bool(re.search('访问异常|验证码|访问验证|安全验证',tree.xpath('string(//title)')))
def pace(host):
 with guard:
  at=max(time.monotonic(),next_start.get(host,0));next_start[host]=at+0.75
 if at>time.monotonic():time.sleep(at-time.monotonic())

def catalog(job):
 query,page=job;url='https://www.foodaily.com/search/articles?'+urllib.parse.urlencode({'search':query,'page':page})
 key='foodaily-'+b.digest(url.encode())[:18];pace('www.foodaily.com')
 raw,r=b.fetch(url,key,WORK/'search');tree=html.fromstring(raw.decode('utf8'))
 if challenge(raw,r,tree):state['hostStops']['www.foodaily.com']={'url':url,'reason':'CATALOG_CHALLENGE'};raise ValueError('CATALOG_CHALLENGE')
 assert r['httpStatus']==200,'CATALOG_HTTP_ERROR'
 rows={}
 for a in tree.xpath('//a[@href]'):
  u=a.get('href');title=' '.join(a.text_content().split())
  if re.fullmatch(r'https://www\.foodaily\.com/articles/\d+',u or '') and title and NOODLE.search(title):
   rows[u]={'url':u,'title':title,'discoveryKind':'publisher_catalog','query':query,'page':page,'discoveryReceipt':str((WORK/'search'/(key+'.http.json')).relative_to(ROOT))}
 b.save(WORK/'search'/(key+'.results.json'),{'query':query,'page':page,'receipt':r,'results':list(rows.values())})
 return list(rows.values())

def linked_posts(raw,receipt,origin):
 text,encoding=decode(raw);tree=html.fromstring(text);rows={}
 for a in tree.xpath('//a[@href]'):
  u=original_post(a.get('href'));title=' '.join(a.text_content().split())
  if u and len(title)>7 and NOODLE.search(title):
   rows[u]={'url':u,'title':title[:300],'discoveryKind':'article_related_link','discoveredAtSource':receipt['finalUrl'],'discoveryReceipt':origin}
 return rows

def collect(candidate):
 u=candidate['url'];host=urllib.parse.urlparse(u).hostname
 if host in state['hostStops']:raise ValueError('HOST_STOPPED_AFTER_CHALLENGE')
 key=b.digest(u.encode())[:24];cached=WORK/'sources'/(key+'.source.json')
 if cached.exists():return json.loads(cached.read_bytes())
 pace(host);raw,r=b.fetch(u,key,WORK/'sources');text,encoding=decode(raw);tree=html.fromstring(text)
 if challenge(raw,r,tree):
  state['hostStops'][host]={'url':u,'httpStatus':r['httpStatus'],'reason':'ACCESS_CHALLENGE'};raise ValueError('ACCESS_CHALLENGE')
 if r['httpStatus']!=200:raise ValueError('HTTP_'+str(r['httpStatus']))
 if 'post.smzdm.com'==host:
  nodes=tree.xpath('//article');assert len(nodes)==1,'SINGLE_ORIGINAL_ARTICLE_REQUIRED'
  selector='//article';body=b.paragraphs(nodes[0]);title=' '.join(tree.xpath('string(//article//h1)').split()) or tree.xpath('string(//title)').strip()
  dates=tree.xpath('//meta[@property="article:published_time"]/@content')
  if not dates:
   found=re.search(r'\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}(?::\d{2})?',nodes[0].text_content()[:1200]);dates=[found[0]] if found else []
  published=dates[0] if dates else None;pid=re.search(r'/(?:talk/)?p/([a-z0-9]+)/',u)[1];platform='smzdm'
 else:
  title,body,selector,published,encoding=industry.extract(raw,r);platform='foodaily';pid=urllib.parse.urlparse(u).path.split('/')[-1]
 if len(body.removeprefix(title).strip())<100:raise ValueError('INSUFFICIENT_ACTUAL_BODY')
 if len(body)>65000:raise ValueError('BODY_REQUIRES_MANUAL_EXTRACTION')
 if not NOODLE.search(body):raise ValueError('NO_ACTUAL_PACKAGED_NOODLE_CONTENT')
 if not published:raise ValueError('ORIGINAL_DATE_REQUIRED')
 date=re.search(r'(\d{4})[-年/.](\d{1,2})[-月/.](\d{1,2})',published)
 if not date:raise ValueError('UNPARSED_DATE')
 day='-'.join([date[1],date[2].zfill(2),date[3].zfill(2)])
 if day>datetime.now().strftime('%Y-%m-%d'):raise ValueError('FUTURE_DATE')
 row={'entity':'social','id':platform+':'+pid,'postId':pid,'platform':platform,'url':u,'title':title,'body':body,'bodySha256':b.digest(body.encode()),
      'publishedAtRaw':published,'sourceEncoding':encoding,'bodyExtractionSelector':selector,'httpReceipt':r,'collectionDirectory':'sources',
      'searchDiscovery':candidate,'topics':[],'sourceType':'unreviewed_public_article','limits':[]}
 b.save(cached,row);return row

def main():
 inventory=json.loads((ROOT/'work/library-cleanup-20261111/round-02/inventory-after-result.json').read_bytes())['body']['result']
 known={canonical(r.get('url') or '') for r in inventory['social']}
 latest=json.loads((ROOT/'work/database-enrichment-20261110/batches/five-topics-multisource-20261111-01/reviewed-source-index.json').read_bytes())
 known.update(canonical(r['url']) for r in latest if r['entity']=='social')
 candidates={};archives=[]
 for directory in [ROOT/'work/five-topic-research-20261111/sources',ROOT/'work/five-topic-collection-20261110/public-sources',ROOT/'work/five-topic-collection-20261110/supplement-01/public-sources',ROOT/'work/five-topic-collection-20261110/supplement-02/sources']:
  for f in directory.glob('*.http.json'):
   receipt=json.loads(f.read_bytes())
   if receipt['httpStatus']!=200 or 'post.smzdm.com/' not in receipt['finalUrl']:continue
   raw=(f.parent/receipt['responseFile']).read_bytes();assert b.digest(raw)==receipt['sha256']
   for u,c in linked_posts(raw,receipt,str(f.relative_to(ROOT))).items():
    if canonical(u) not in known:candidates.setdefault(u,c)
 state['archivedLinkCandidates']=len(candidates);persist()
 jobs=[('方便面',p) for p in range(1,16)]+[(q,p) for q in ['拌面','拉面','冻干','非油炸','菌菇','五谷道场','汤达人','食药同源'] for p in range(1,3)]
 for job in jobs:
  if 'www.foodaily.com' in state['hostStops']:break
  try:
   rows=catalog(job);state['searches'].append({'query':job[0],'page':job[1],'found':len(rows)})
   for c in rows:
    if canonical(c['url']) not in known:candidates.setdefault(c['url'],c)
  except Exception as e:state['rejected'].append({'catalog':job,'reason':str(e)})
  persist();print(json.dumps({'stage':'discovery','catalogs':len(state['searches']),'candidates':len(candidates)}),flush=True)
 b.save(WORK/'discovered-candidates.json',list(candidates.values()))
 # Bound this run while collecting enough distinct original pages to review.
 selected=list(candidates.values())[:360]
 with ThreadPoolExecutor(max_workers=2) as pool:
  tasks={pool.submit(collect,c):c for c in selected}
  for f in as_completed(tasks):
   try:state['sources'].append(f.result())
   except Exception as e:state['rejected'].append({'url':tasks[f]['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-semantic-review',finishedAt=b.now(),uniqueSources=len({r['id'] for r in state['sources']}));persist()
 print(json.dumps({'stage':'completed','sources':len(state['sources']),'hostStops':state['hostStops']}),flush=True)

if __name__=='__main__':main()