"""Collect readable public source pages, keeping search results out of evidence counts.""" from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone import hashlib import json from pathlib import Path import re import threading import urllib.error import urllib.parse import urllib.request from lxml import html WORK = Path(__file__).resolve().parents[1] / 'work/five-topic-collection-20261110' PLAN = json.loads((WORK / 'collection-plan.json').read_bytes()) PUBLIC = WORK / 'public-sources' PUBLIC.mkdir(exist_ok=True) SEARCH = WORK / 'public-search' QUERIES = { 'wellness': ['方便面 中式养生 当归', '方便面 药食同源', '菌菇汤 方便面 非油炸', '养生方便面 site:thepaper.cn', '方便面 菌菇 测评 site:post.smzdm.com'], 'student': ['大学生 宿舍 泡面 测评', '学生 夜宵 方便面', '宿舍 方便面 不用锅', '宿舍 泡面 site:post.smzdm.com', '方便面 夜宵 site:thepaper.cn'], 'premium': ['高端 方便面 真实肉块', '方便面 升级 满汉大餐', '方便面 中产 拉面说', '高端 方便面 site:thepaper.cn', '拉面说 满汉大餐 测评 site:post.smzdm.com'], 'stockpile': ['方便面 囤货 测评', '家庭 囤货 方便面 保质期', '方便面 整箱 组合装', '方便面 囤货 site:post.smzdm.com', '方便面 囤货 保质期 site:sohu.com'], 'novelty': ['方便面 联名 新品', '方便面 新口味 测评', '泡面 尝鲜 冬阴功 香菜', '方便面 联名 site:thepaper.cn', '方便面 新口味 测评 site:post.smzdm.com'], } NOODLES = re.compile(r'方便面|泡面|即食面|速食面|杯面|桶面|碗面|方便拌面|速食拌面|火鸡面') PLATFORMS = {'sohu.com': 'sohu', 'thepaper.cn': 'thepaper', 'smzdm.com': 'smzdm', 'zhihu.com': 'zhihu', 'weixin.qq.com': 'wechat', 'xinhuanet.com': 'xinhua', 'news.cn': 'xinhua', 'people.com.cn': 'people', 'chinanews.com.cn': 'chinanews', 'chinanews.com': 'chinanews', 'cctv.com': 'cctv', 'foodaily.com': 'foodaily', 'foodtalks.cn': 'foodtalks', 'cbndata.com': 'cbndata', 'jiemian.com': 'jiemian', 'stcn.com': 'stcn', 'yicai.com': 'yicai', 'cnr.cn': 'cnr', 'sina.com.cn': 'sina', '163.com': 'netease', 'news.qq.com': 'tencent'} STATE = {'schema': 'ksf.five-topic.public-collection.v1', 'status': 'running', 'startedAt': None, 'searches': [], 'sources': [], 'rejected': [], 'productionWrites': 0} GUARD = threading.Lock() def now(): return datetime.now(timezone.utc).isoformat() def digest(raw): return hashlib.sha256(raw).hexdigest() def save(path, data): temporary = path.with_suffix(path.suffix + '.next') temporary.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding='utf8') temporary.replace(path) def persist(): STATE['updatedAt'] = now() save(WORK / 'public-collection-result.json', STATE) def platform(url): host = urllib.parse.urlparse(url).hostname or '' return next((value for suffix, value in PLATFORMS.items() if host == suffix or host.endswith('.' + suffix)), None) def fetch(url, name, target): previous = target / (name + '.http.json') if previous.exists(): receipt = json.loads(previous.read_bytes()) body = target / receipt['responseFile'] if receipt['requestedUrl'] == url and body.exists() and digest(body.read_bytes()) == receipt['sha256']: return body.read_bytes(), receipt request = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) try: with urllib.request.urlopen(request, timeout=25) as response: raw = response.read(6000001) status, final, content_type = response.status, response.url, response.headers.get('Content-Type', '') except urllib.error.HTTPError as error: raw, status, final, content_type = error.read(6000001), error.code, error.url, error.headers.get('Content-Type', '') if len(raw) > 6000000: raise ValueError('SOURCE_TOO_LARGE') body = name + '.body' (target / body).write_bytes(raw) receipt = {'requestedUrl': url, 'finalUrl': final, 'httpStatus': status, 'contentType': content_type, 'collectedAt': now(), 'responseFile': body, 'bytes': len(raw), 'sha256': digest(raw)} save(target / (name + '.http.json'), receipt) return raw, receipt def search(topic, index, query): key = topic + '-sogou-' + str(index + 1) file = SEARCH / (key + '.results.json') if file.exists(): return json.loads(file.read_bytes()) url = 'https://www.sogou.com/web?' + urllib.parse.urlencode({'query': query}) raw, receipt = fetch(url, key, SEARCH) rows = [] if receipt['httpStatus'] == 200: root = html.fromstring(raw) for heading in root.xpath('//h3'): links = heading.xpath('.//a[@href]') if not links: continue title = re.sub(r'\s+', ' ', heading.text_content()).strip() # Search snippets alone never count as source bodies. if not title or len(title) < 6: continue rows.append({'title': title, 'url': urllib.parse.urljoin(receipt['finalUrl'], links[0].get('href')), 'topic': topic, 'query': query, 'searchReceipt': key + '.http.json'}) result = {'topic': topic, 'query': query, 'receipt': receipt, 'results': rows} save(file, result) return result def resolve(raw, receipt, key): for step in range(2): host = urllib.parse.urlparse(receipt['finalUrl']).hostname or '' if not host.endswith('sogou.com'): return raw, receipt root = html.fromstring(raw) urls = root.xpath('//meta[translate(@http-equiv,"REFSH","refsh")="refresh"]/@content') target = None for value in urls: found = re.search(r'url\s*=\s*[\'\"]?(https?://[^\'\"\s>]+)', value, re.I) if found: target = found.group(1) if not target: text = raw.decode('utf8', errors='replace') found = re.search(r'(?:location\.(?:replace|assign)\s*\(|(?:window\.)?location(?:\.href)?\s*=\s*)[\'\"](https?://[^\'\"\s]+)', text) if found: target = found.group(1).replace('\\/', '/') if not target: raise ValueError('SEARCH_REDIRECT_NOT_ORIGINAL_PAGE') raw, receipt = fetch(target, key + '-redirect-' + str(step + 1), PUBLIC) return raw, receipt def paragraphs(node): clone = html.fromstring(html.tostring(node, encoding='unicode')) for element in clone.xpath('.//script|.//style|.//noscript|.//nav|.//footer|.//form|.//button|.//*[contains(concat(" ",normalize-space(@class)," ")," recommend-tab ")]'): element.drop_tree() nodes = clone.xpath('.//p|.//h1|.//h2|.//h3|.//blockquote|.//li') chunks = [re.sub(r'\s+', ' ', element.text_content()).strip() for element in nodes] chunks = [part for part in chunks if part] if not chunks: chunks = [re.sub(r'\s+', ' ', clone.text_content()).strip()] return '\n'.join(dict.fromkeys(chunks)) def extract(raw, receipt): declared = re.search(r'charset\s*=\s*([^;\s]+)', receipt.get('contentType', ''), re.I) encoding = declared.group(1).strip('"\'') if declared else 'utf8' try: decoded = raw.decode(encoding) except (LookupError, UnicodeDecodeError): try: decoded = raw.decode('utf8'); encoding = 'utf8' except UnicodeDecodeError: decoded = raw.decode('gb18030'); encoding = 'gb18030' root = html.fromstring(decoded) # Never treat a verification/login response as an article. title = ''.join(root.xpath('//meta[@property="og:title"]/@content')[:1]) or root.xpath('string(//title)') title = re.sub(r'\s+', ' ', title).strip() if re.search(r'安全验证|访问验证|验证码|系统繁忙|访问异常|扫码登录|页面不存在|404|没有找到', title, re.I): raise ValueError('VERIFICATION_OR_ERROR_PAGE') selectors = [ '//*[@id="js_content"]', '//*[@id="artibody"]', '//*[contains(concat(" ",normalize-space(@class)," ")," article-content ")]', '//*[contains(concat(" ",normalize-space(@class)," ")," article_content ")]', '//*[contains(concat(" ",normalize-space(@class)," ")," article-main ")]', '//*[@id="detail_content"]', '//*[@id="article-content"]', '//*[contains(concat(" ",normalize-space(@class)," ")," rich_media_content ")]', '//article', '//*[@itemprop="articleBody"]', '//*[contains(concat(" ",normalize-space(@class)," ")," article_content_wrapper ")]', '//*[contains(concat(" ",normalize-space(@class)," ")," post_text ")]', '//*[contains(concat(" ",normalize-space(@class)," ")," entry-content ")]', ] body, selector = None, None for candidate in selectors: texts = [paragraphs(node) for node in root.xpath(candidate)] texts = [value for value in texts if len(value) >= 300] if texts: body, selector = max(texts, key=len), candidate; break if not body: raise ValueError('ARTICLE_BODY_NOT_LOCATED') if not NOODLES.search(title + '\n' + body): raise ValueError('NOT_PACKAGED_NOODLE_CONTENT') date_values = root.xpath('//meta[@property="article:published_time" or @name="pubdate" or @name="publishdate" or @name="publish-date" or @name="PublishDate" or @name="weibo:article:create_at"]/@content') dates = [value for value in date_values if re.search(r'\d{4}[-/]\d{1,2}[-/]\d{1,2}', value)] if not dates: headers = root.xpath('//article') header = headers[0].text_content()[:1100] if headers else '' visible = re.search(r'\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}(?::\d{2})?', header) if visible: dates = [visible.group(0)] return title, body, selector, dates[0] if dates else None, encoding def collect(candidate): key = digest(candidate['url'].encode())[:24] record_file = PUBLIC / (key + '.source.json') if record_file.exists(): return json.loads(record_file.read_bytes()) raw, receipt = fetch(candidate['url'], key, PUBLIC) if receipt['httpStatus'] != 200: raise ValueError('HTTP_' + str(receipt['httpStatus'])) raw, receipt = resolve(raw, receipt, key) site = platform(receipt['finalUrl']) if not site: raise ValueError('UNSUPPORTED_OR_SEARCH_RESULT_HOST') title, body, selector, published, encoding = extract(raw, receipt) source_topics = [] for topic in PLAN['topics']: hit = [term for term in topic['keywords'] if term in title + '\n' + body] if not hit: continue passages = [] for term in hit: start = body.find(term) if start >= 0: passages.append(body[max(0, start - 65):start + 150]) source_topics.append({'topic': topic['id'], 'matchedTerms': hit, 'passages': passages[:5]}) if not source_topics: raise ValueError('NO_REQUESTED_TOPIC_IN_SOURCE_BODY') # Drop referrer/tracking fields from the identity; preserve the exact URL in the receipt. parsed = urllib.parse.urlparse(receipt['finalUrl']) query = urllib.parse.parse_qs(parsed.query) stable_query = urllib.parse.urlencode({k: query[k][0] for k in ['__biz', 'mid', 'idx'] if k in query}) canonical = urllib.parse.urlunparse((parsed.scheme, parsed.netloc, parsed.path, '', stable_query, '')) record = {'schema': 'ksf.five-topic.public-source.v1', 'sourceType': 'public_article_not_ecommerce_review', 'sourceId': 'url-sha256:' + digest(canonical.encode()), 'identityBasis': 'canonical_original_url_sha256', 'platform': site, 'title': title, 'url': canonical, 'originalResolvedUrl': receipt['finalUrl'], 'publishedAtRaw': published, 'collectedAt': receipt['collectedAt'], 'body': body, 'bodySha256': digest(body.encode()), 'bodyExtractionSelector': selector, 'sourceEncoding': encoding, 'httpReceipt': receipt, 'searchDiscovery': candidate, 'topics': source_topics, 'verification': 'readable_original_article_with_preserved_http_bytes', 'publishedToDatabase': False, 'limits': ['原作者/媒体观点，不当作电商购买评价', '文中健康或市场数字未在此轮独立验证', '保留原文字段与原件，不将采集时刻写成发布时间']} save(record_file, record) return record def main(): STATE['startedAt'] = now() candidates = {} for topic, queries in QUERIES.items(): for index, query in enumerate(queries): try: result = search(topic, index, query) STATE['searches'].append({'topic': topic, 'query': query, 'candidateCount': len(result['results']), 'sha256': result['receipt']['sha256']}) # Prefer text articles and actual experiences, excluding obvious recipe/video/category pages. filtered = [row for row in result['results'] if not re.search(r'做法步骤|菜谱大全|怎么做好吃|视频大全|中毒|健康知识|价格表|批发价格', row['title'])] for row in filtered[:10]: candidates.setdefault(row['url'], row) except Exception as error: STATE['searches'].append({'topic': topic, 'query': query, 'error': type(error).__name__}) persist() print(json.dumps({'stage': 'search', 'topic': topic, 'query': query, 'candidates': len(candidates)}, ensure_ascii=False), flush=True) with ThreadPoolExecutor(max_workers=3) as pool: tasks = {pool.submit(collect, row): row for row in candidates.values()} for task in as_completed(tasks): candidate = tasks[task] try: record = task.result() with GUARD: STATE['sources'].append(record) except Exception as error: with GUARD: STATE['rejected'].append({'candidate': candidate, 'reason': str(error) if isinstance(error, ValueError) else type(error).__name__}) with GUARD: persist() print(json.dumps({'stage': 'body', 'accepted': len(STATE['sources']), 'rejected': len(STATE['rejected']), 'candidates': len(candidates)}, ensure_ascii=False), flush=True) unique = {} for row in STATE['sources']: key = row['sourceId'] if key in unique: known = {item['topic'] for item in unique[key]['topics']} unique[key]['topics'].extend(item for item in row['topics'] if item['topic'] not in known) else: unique[key] = row STATE['sources'] = list(unique.values()) STATE['status'] = 'complete' STATE['finishedAt'] = now() STATE['uniqueSources'] = len(STATE['sources']) STATE['countsByTopic'] = {topic['id']: sum(any(item['topic'] == topic['id'] for item in row['topics']) for row in STATE['sources']) for topic in PLAN['topics']} STATE['databaseImported'] = 0 persist() print(json.dumps({'status': STATE['status'], 'uniqueSources': STATE['uniqueSources'], 'countsByTopic': STATE['countsByTopic'], 'databaseImported': 0}, ensure_ascii=False), flush=True) if __name__ == '__main__': main()