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

"""Target gaps with readable original community pages; never publish search hits."""
from concurrent.futures import ThreadPoolExecutor, as_completed
import importlib.util
import json
from pathlib import Path
import re
import urllib.parse
from lxml import html

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('multi', ROOT/'scripts/collect-five-topic-multisource.py')
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
WORK = m.WORK
QUERIES = [
 ('student', 'site:post.smzdm.com 宿舍 泡面 测评'),
 ('student', 'site:post.smzdm.com 夜宵 拌面 实测'),
 ('student', 'site:sspai.com 方便面 测评'),
 ('student', 'site:post.smzdm.com 学生 方便面 开水'),
 ('novelty', 'site:post.smzdm.com 香菜 方便面 测评'),
 ('novelty', 'site:post.smzdm.com 冬阴功 泡面 试吃'),
 ('novelty', 'site:post.smzdm.com 藤椒 拌面 测评'),
 ('stockpile', 'site:post.smzdm.com 囤货 方便面 保质期'),
 ('wellness', 'site:post.smzdm.com 菌菇 非油炸 方便面'),
]

def search(job):
 i,(topic,q) = job
 url='https://www.so.com/s?'+urllib.parse.urlencode({'q':q})
 raw,r=m.base.fetch(url,'focused-'+str(i),m.SEARCH)
 tree=html.fromstring(raw.decode('utf8',errors='replace'))
 if r['httpStatus']!=200 or re.search('验证码|访问验证',tree.xpath('string(//title)')):
  raise ValueError('SEARCH_UNAVAILABLE')
 rows={}
 for a in tree.xpath('//a[@data-mdurl]'):
  u=a.get('data-mdurl'); title=m.text(a)
  if not u or not re.search('方便面|泡面|拌面|拉面|火鸡面|合味道|汤达人',title):continue
  if not re.match(r'https://(?:post\.smzdm\.com/(?:talk/)?p/|sspai\.com/post/)',u):continue
  rows[u]={'url':u,'title':title,'topic':topic,'query':q,'searchReceipt':'focused-'+str(i)+'.http.json'}
 m.base.save(m.SEARCH/('focused-'+str(i)+'.results.json'),{'query':q,'httpReceipt':r,'results':list(rows.values())})
 return list(rows.values())

def main():
 prior=json.loads((ROOT/'work/library-cleanup-20261111/round-02/inventory-after-result.json').read_bytes())['body']['result']
 known={m.canonical(r.get('url') or '') for r in prior['social']}
 state={'status':'running','searches':[],'sources':[],'rejected':[]};candidates={}
 with ThreadPoolExecutor(max_workers=3) as pool:
  tasks={pool.submit(search,j):j for j in enumerate(QUERIES,1)}
  for f in as_completed(tasks):
   try:
    rows=f.result();state['searches'].append({'query':tasks[f][1][1],'found':len(rows)})
    for r in rows:
     if m.canonical(r['url']) not in known:candidates.setdefault(r['url'],r)
   except Exception as e:state['rejected'].append({'job':tasks[f],'reason':str(e)})
 print(json.dumps({'candidates':len(candidates)}),flush=True)
 with ThreadPoolExecutor(max_workers=2) as pool:
  tasks={pool.submit(m.collect,r):r for r in candidates.values()}
  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)})
   m.base.save(WORK/'focused-collection-result.json',state)
   print(json.dumps({'readable':len(state['sources']),'rejected':len(state['rejected'])}),flush=True)
 state.update(status='awaiting-source-review',candidateCount=len(candidates));m.base.save(WORK/'focused-collection-result.json',state)

if __name__=='__main__':main()