#!/usr/bin/env python3 """ 统一搜索服务（端口8000） 集成：gov FTS5搜索 + 中项网/中能联合/中策大数据 """ import http.server, sqlite3, urllib.parse, re, os, html, json, datetime, io, csv, signal, sys, markdown, uuid, string def contains_chinese(text): return bool(re.search(r"[\u4e00-\u9fff\u3400-\u4dbf]", text)) DB_PATH = "/root/search.db" CONTACT_DB_PATH = "/root/ccpc.db" ZNLH_DB_PATH = "/home/ccbuild/znlh.db" ZC_DB_PATH = "/root/ZC.db" EIA_DB_PATH = "/root/eia.db" CSV_PATH = "/root/data.csv" PORT = 8000 def extract_investment(text): if not text: return '-' m = re.search(r'[总]?投资[额金]?[：:为]?\s*[约]?\s*[\d,.]+\s*[万亿千百十万千百元美]?[元美]?', text) if m: return m.group(0).strip() m = re.search(r'投资.{0,6}?[\d,.]+[万亿]?', text) if m: return m.group(0).strip() return '-' PHASES = [ {'id':'eia','label':'环评阶段','keywords':'环评 OR 环境影响评价 OR 环境影响报告 OR 环境影响登记 OR 第一次公示 OR 第二次公示 OR 征求意见 OR 公众参与'}, {'id':'approval','label':'审批报批','keywords':'审批 OR 报批 OR 批复 OR 核准 OR 备案 OR 行政许可 OR 审查意见 OR 同意建设'}, {'id':'bidding','label':'招标采购','keywords':'招标 OR 中标 OR 采购 OR 比选 OR 邀标 OR 竞标 OR 竞争性谈判 OR 询价 OR 招标公告 OR 中标候选人'}, {'id':'acceptance','label':'竣工验收','keywords':'验收 OR 竣工 OR 试运行 OR 投产 OR 调试 OR 环境保护验收 OR 竣工验收'}, {'id':'planning','label':'规划设计','keywords':'规划 OR 设计 OR 可行性研究 OR 可研 OR 选址 OR 用地预审 OR 方案设计 OR 初步设计'}, {'id':'security','label':'安全评价','keywords':'安全评价 OR 安全设施 OR 安全验收 OR 安全预评价 OR 安全专篇 OR 安全生产 OR 风险评估'}, ] PHASE_MAP = {p['id']: p for p in PHASES} # ─── 行业分类（一级标签，仅用标题匹配）─── INDUSTRY_RULES = [ ('pharma', '制药与生物技术', ['APIs','API','GMP','cGMP','疫苗','生物制品','干细胞','抗体','基因','诊断试剂','医疗器械', '药品', '制药', '医药', '药业','医药','胶原蛋白','单克隆','血浆','胰岛素','造影剂','缝合线','导管','内窥镜','起搏器','透析','神经外科','片剂','丸剂','胶囊','软膏','注射剂','注射器','口服液','辅料','CDMO','多肽', '原药', '母药', '原料药', '兽药', '制剂', '中成药', '生物医药', '制药项目', '医药中间体', '原料药及', '药厂', '制药厂', '医药产业园', '药用']), ('production', '生产/开采', ['油田','气田','采油','采气','钻井','页岩气','海上平台','酸气处理','天然气处理站','天然气集气站','天然气生产平台','煤层气生产平台']), ('transmission', '管道输送', ['管道','输气','输油','门站','计量站','压气站','阀室','装卸站及专用铁路']), ('terminals', '终端/接收站', ['接收站','LNG终端','LNG','天然气终端','天然气储存终端','油库','罐区','原油储罐','成品油罐','储运','成品油仓储码头','电煤港','煤炭港', '油品码头', '原油码头', '成品油码头', 'LNG码头', '液化天然气码头', '天然气码头', '油气码头', '液体化工码头', '加油站', '加气站', '加油站扩建', '成品油销售', '燃气公司', '燃气供应', '液化气站', '充电站', '换电站']), ('port', '港口码头', ['港口','码头','泊位','港区','港务','集装箱','散货','件杂货','客滚','轮渡','航运','疏浚','航道','船闸','引航','渡口','装卸']), ('hpi', '石油炼制', ['炼化一体','炼化','炼油','炼厂','石脑油','催化裂化','加氢裂化','FCC']), ('altfuel', '替代燃料', ['生物柴油','生物乙醇','可持续航空燃料','沼气','燃料电池','乙醇燃料','甲醇燃料','SAF', '生物质颗粒', '生物质燃料', '生物质成型', '生物质能', '生物质气化']), ('power', '电力', ['发电','电站','光伏','风电','太阳能','储能','BESS','CAES','CSP','变电站','变电所','输变电','输电','电网','核电','水电站','抽水蓄能','压缩空气储能','煤电','供热', '生物质发电', '生物质锅炉', '生物质热电']), ('metals', '金属与矿物', ['金属与矿物','有色金属','稀土','稀有金属','不锈钢','钢铁','冶金','矿山','采矿','选矿','尾矿','冶炼','钢厂','轧钢','炼铁','炼钢','铁合金','钢铁厂','金矿','铁矿','铜矿','铝土矿','磷矿','石墨矿','石墨化电极','石墨','钢结构','石料','多金属矿','锌矿','铅矿','镍矿','镁合金','铝合金','钛合金','铜合金','锌合金','轻合金','铝板','铝业','铜业','锌业','金属','矿物','铝','铜','锌','铅','镍','锡','钨','钼','合金','矿','水泥','预拌混凝土','建材','石材','瓷砖','石膏板','防水材料','保温材料','耐火材料','马口铁','易拉罐', '混凝土', '砂浆', '商砼', '沥青混凝土', '水泥制品', '搅拌站', '预拌砂浆']), ('pulp', '制浆/造纸/木材', ['纸浆','造纸','纸板','木材','木业','人造板','瓦楞','刨花板','胶合板','纸厂','印刷品', '印刷包装','纸包装']), ('food', '食品与饮料', ['食品','饮料','肉类','屠宰','乳','牛奶','果汁','啤酒','调味','面包','豆制品','食用油','饲料','糖','淀粉','酱油','酵母','茶','咖啡','水产','禽','畜','碾米','酒厂','烈酒','植物产品','瓶装水','矿泉水','烘焙','花生加工','鸡蛋加工','鱼片','香料', '养殖', '生猪', '蛋鸡', '肉鸭', '肉鸡', '水产养殖', '酿酒', '白酒', '啤酒厂', '酱', '预制菜', '果干', '果汁厂', '肉类加工', '屠宰场', '饲料厂']), ('cpi', '化工加工', ['废矿物油','化工','化学品','树脂','聚合物','聚乙烯','聚丙烯','PVC','PET','PTA','ABS','EVA','HDPE','LDPE','PE管','PPR','塑料','橡胶','涂料','染料','肥料','农药','溶剂','催化剂','酸','酯','醇','硫酸','盐酸','氢氧化','氧化物','丙烯','乙烯','苯','酮','醚','胺','烯','硅','氟','氯','溴','碘','粘合剂','添加剂','阻燃剂','表面活性剂','精细化学品','水处理剂','复合肥','水溶肥','颜料','洗涤剂','增塑剂','活性炭','高纯度化学品','湿化学品','过氧化氢','甲醛','甲醇','氨','氧气','空气分离','烧碱','化学试剂','化学中间体','炭黑','气凝胶','脱硫剂','纺织助剂','纤维素纤维','紫外线吸收剂','电解质','杀菌剂','杀虫剂','除草剂','制冷剂','减水剂','乙炔','双酚','莱赛尔纤维','脂肪','炔','腈','酐','醛','酚','氢能','绿氢','氢气','电解水制氢','电解制氢','水电解制氢','制氢','氟化石墨','全氟聚醚','氟化改性','高纯氟气','软包装','复合膜', '植物保护', '种子处理剂', '农药制剂', '农药中间体', '植物蛋白', '生物农药', '农药原药']), ('logistics', '物流', ['物流','物流中心','物流园','仓储物流','保税物流','冷链物流','快递分拨']), ('manufacturing', '工业制造', ['建筑用砂','制造','设备','机械','零部件','电子','汽车','半导体','电池','家电','仪器','配件','泵','压缩机','阀门','轴承','传感器','显示屏','印刷电路板', '印刷线路板','连接器','变压器','无人机','模具','晶圆','数控机床','造船','起重机','空调','触摸屏','芯片','机器人','雷达','集成电路','光刻胶','电机','锅炉','过滤器','船厂','机床','紧固件','风机','飞机','数据中心','家具','包装', '电缆', '电线电缆', '特种电缆', '面料', '染整', '织造', '针织', '梭织', '制鞋', '鞋', '标签', '不干胶', '五金', '五金制品', '制品', '电子元件', '光学']), ] INDUSTRY_MAP = {ind_id: {'id': ind_id, 'label': label, 'kw': kws} for ind_id, label, kws in INDUSTRY_RULES} # 预计算关键词对（按长度降序），避免 classify_industry 每次调用重复构建+排序 CLASSIFY_PAIRS = sorted( ((len(kw), ind_id, kw.lower()) for ind_id, label, kws in INDUSTRY_RULES for kw in kws), key=lambda x: -x[0], ) # 行政公告排除（2026-08-11 用户决策: 行政公告保持 other） ADMIN_EXCLUDE = [ '医保', '药店', '医药机构', '定点零售', '零售药店', '医保服务协议', '医保定点', '集采药品销售', '药品经营', '药品价格', '门诊慢性病', '医疗保障', '医保目录', '药师', '执业药师', ] def classify_industry(title): """仅用标题做一级行业分类，返回行业 id。最长关键词优先（用户体系）。 小写匹配：英文关键词大小写不敏感（BESS/bess/Bess 均命中）。""" if not title: return 'other' t = title.lower() # 行政公告排除: 医保/药店/定点零售等 → other (用户决策 2026-08-11) for kw in ADMIN_EXCLUDE: if kw.lower() in t: return 'other' # 地名等误伤排除（命中即不是 metals） if '铜梁' in t or '铜仁' in t: t = t.replace('铜梁', '·').replace('铜仁', '·') # 用户2026-08-23: 包含'制品' → 制造业, 除非命中已有的行业特定制品词(生物制品/水泥制品/豆制品/五金制品等) if '制品' in t: existing_goods = [kw for ind_id, label, kws in INDUSTRY_RULES for kw in kws if kw.endswith('制品') and len(kw) >= 3] if not any(kw in t for kw in existing_goods): return 'manufacturing' for _len, ind_id, kw in CLASSIFY_PAIRS: if kw in t: return ind_id return 'other' def ensure_industry_column(): """启动时确保 gov_raw 有 industry 列""" try: conn = sqlite3.connect(DB_PATH, timeout=10) conn.execute("ALTER TABLE gov_raw ADD COLUMN industry TEXT DEFAULT 'other'") conn.commit() conn.close() except Exception: pass # 列已存在则忽略 CATEGORIES = [ {'id': '资源网站', 'label': '资源网站'}, ] def esc(s): if s is None: return '' return str(s).replace('&','&').replace('<','<').replace('>','>').replace('"','"') def render_md(text): """Render Markdown, ensuring blank lines before tables for proper table detection""" if not text: return '' # Ensure blank line before table: a line containing ' | ' followed by line starting with '---' text = re.sub(r'(? 1: exc.append(p[1:]) else: inc.append(p) return inc, exc # ─── Gov DB 部分 ─── def normalize_query(q): """Replace punctuation (Chinese + English) with spaces. 保留 - : ( ) （ ） 作为关键词一部分（前面无空格时视为普通字符，避免 FTS 高级语法误触发）。""" if not q: return q punct_set = set(string.punctuation.replace("-", "").replace(":", "").replace("(", "").replace(")", "") + "，。、；？！…—·～「」『』〖〗【】〔〕〈〉《》") result = "".join(" " if c in punct_set else c for c in q) result = re.sub(r"\s+", " ", result).strip() return result def sanitize_fts_query(q): """Remove FTS5 special chars and replace punctuation with spaces. FTS5 特殊字符: ( ) * ^ \" [ ] . / - : 都替换为空格，避免语法错误（no such column / syntax error）。""" if not q: return q q = normalize_query(q) q = re.sub(r'[()（）*^"\[\]./:\-]', ' ', q) q = re.sub(r'\s+', ' ', q).strip() return q def get_db(): conn = sqlite3.connect(DB_PATH, timeout=10) conn.row_factory = sqlite3.Row conn.execute("PRAGMA busy_timeout=5000") conn.execute('PRAGMA journal_mode=WAL') return conn # ===== originals 标题匹配（项目ID展示）===== _ORIGINALS_DB = '/root/originals.db' _ORIGINALS_IDX = None # (proj_titles, comp_titles) 各为 [(title, id_code), ...] _ORIGINALS_IDX_MT = 0.0 # 索引对应的 originals.db mtime def _clean_otitle(t): """清洗 originals 标题: 去 tab 前缀 / 数字前缀 / 尾标点""" if not t: return '' t = t.strip() if '\t' in t: t = t.split('\t')[-1].strip() t = re.sub(r'^\d{6,10}\s+', '', t) t = re.sub(r'[，。、；：\s]+$', '', t) return t _OPROJ_KEY = re.compile(r'(项目|工程|装置|年产|万吨|扩建|技改|新建|生产线|车间)') def _load_originals_index(): """懒加载 originals.db 标题索引, mtime 变化时重建. 返回 (proj_titles, comp_titles)""" global _ORIGINALS_IDX, _ORIGINALS_IDX_MT try: mt = os.path.getmtime(_ORIGINALS_DB) except OSError: return _ORIGINALS_IDX or ([], []) if _ORIGINALS_IDX is not None and mt == _ORIGINALS_IDX_MT: return _ORIGINALS_IDX proj, comp = [], [] seen_p, seen_c = set(), set() if os.path.exists(_ORIGINALS_DB): try: conn = sqlite3.connect(_ORIGINALS_DB, timeout=5) rows = conn.execute('SELECT id_code, title FROM tsk_data').fetchall() conn.close() for code, t in rows: ct = _clean_otitle(t) if len(ct) < 8: continue if len(ct) >= 12 and _OPROJ_KEY.search(ct): if ct not in seen_p: seen_p.add(ct) proj.append((ct, code)) elif 8 <= len(ct) <= 30: if ct not in seen_c: seen_c.add(ct) comp.append((ct, code)) except Exception: pass # 长标题优先: 匹配到最具体的那条 proj.sort(key=lambda x: len(x[0]), reverse=True) comp.sort(key=lambda x: len(x[0]), reverse=True) _ORIGINALS_IDX = (proj, comp) _ORIGINALS_IDX_MT = mt return _ORIGINALS_IDX def find_originals_id(title): """按标题匹配 originals 项目 → (id_code, level), level: 'proj'|'comp'|None""" if not title: return None, None st = title.strip() st = re.sub(r'(环境影响报告书|环评|环境影响评价|公示|受理|审批|批复|决定).*$', '', st) st = re.sub(r'[，。、；：\s]+$', '', st) if len(st) < 10: return None, None proj, comp = _load_originals_index() for octitle, code in proj: if len(octitle) > len(st): continue if octitle in st: return code, 'proj' for octitle, code in comp: if len(octitle) > len(st): continue if octitle in st: return code, 'comp' return None, None def _orig_id_label(id_code): """9位=Project ID, 7位=Plant ID, 其余=None""" s = str(id_code) if id_code is not None else '' if re.fullmatch(r'\d{9}', s): return '项目ID', 'proj' if re.fullmatch(r'\d{7}', s): return '企业ID', 'plant' return None, None def search(query, page=1, page_size=10, sort="relevance", phase=None, category=None, daterange="all", industry=None): """返回 (rows, total, has_more)""" if len(str(query).strip().replace(' ', '')) <= 2: return [], 0, False # 2026-09-03 短词守卫: 不执行检索 conn = sqlite3.connect(DB_PATH, timeout=10) conn.row_factory = sqlite3.Row conn.execute('PRAGMA journal_mode=WAL') c = conn.cursor() params = []; where_parts = [] use_fts = False if query: use_fts = True params.append(query) if phase: kw = PHASE_MAP[phase]['keywords'] if not use_fts: use_fts = True params.append(kw) else: query += ' AND (' + kw + ')' params[0] = query if category: where_parts.append("r.category = ?"); params.append(category) if industry and industry != 'all': where_parts.append("r.industry = ?"); params.append(industry) if daterange and daterange != 'all': days = {'1d':1,'7d':7,'30d':30}[daterange] # 格式防御: 非标准 YYYY-MM-DD 日期(中文格式/空/乱码)不进时间窗口, 避免字符串比较误判污染 where_parts.append(f"(r.publish_date GLOB '20[0-9][0-9]-[0-9][0-9]-[0-9][0-9]*' AND SUBSTR(r.publish_date,1,10) >= date('now', '-{days} days'))") order_sql = 'rank' if sort == 'relevance' else 'r.publish_date DESC' if sort == 'date_desc' else 'r.publish_date ASC' fetch_size = page_size + 1 offset = (page-1)*page_size total = 0 if use_fts: q = params[0] # 标点符号统一视为空格 q = normalize_query(q) params[0] = q # 解析 Google 语法：空格=AND, -keyword=排除 inc_terms, exc_terms = parse_google_query(q) # ── gov_entity 精确实体反查 (2026-09-03): 完整公司名/电话 出现在正文也命中 ── if not exc_terms and len(inc_terms) == 1 and ' ' not in q.strip(): _term = inc_terms[0] _ent = None if contains_chinese(_term) and len(_term) >= 3: _ent = ('company', _term) else: _dig = re.sub(r'\D', '', _term) if 7 <= len(_dig) <= 13: _ent = ('phone', _dig) if _ent: _ew = ' AND ' + ' AND '.join(where_parts) if where_parts else '' try: c.execute(f'SELECT COUNT(*) FROM gov_entity e JOIN gov_raw r ON r.id = e.doc_id WHERE e.etype = ? AND e.value = ?{_ew}', (_ent[0], _ent[1]) + tuple(params[1:])) _et = c.fetchone()[0] if _et > 0: c.execute(f'SELECT r.*, "" AS rank FROM gov_entity e JOIN gov_raw r ON r.id = e.doc_id WHERE e.etype = ? AND e.value = ?{_ew} ORDER BY r.publish_date DESC, r.id DESC LIMIT ? OFFSET ?', (_ent[0], _ent[1]) + tuple(params[1:]) + (fetch_size, offset)) rows = [dict(r) for r in c.fetchall()] has_more = len(rows) > page_size if has_more: rows = rows[:page_size] conn.close() return rows, _et, has_more except sqlite3.OperationalError: pass # 实体表异常 → 回落常规搜索 # 单英文词(域名/URL) → 走 page_url 索引快路径 (LIKE 'http://%kw%' 常量前缀命中 idx_gov_raw_page_url) if len(inc_terms) == 1 and not contains_chinese(q) and not exc_terms: kw = inc_terms[0] # 关键: SQLite LIKE 索引优化只对编译期字面量生效; 参数化 ? 会退化为全表扫(~43s) # 字面量 'http://%kw%' 走 idx_gov_raw_page_url 范围扫描 ~1.5s。 # kw 必须白名单校验(域名/URL 只含字母数字._-), 防注入 if re.fullmatch(r'[A-Za-z0-9._\-]+', kw): url_pat = 'http://%' + kw + '%' url_pat2 = 'https://%' + kw + '%' fts_where_sql = '' if where_parts: fts_where_sql = ' AND ' + ' AND '.join(where_parts) try: c.execute(f"SELECT COUNT(*) FROM gov_raw r WHERE (r.page_url LIKE '{url_pat}' OR r.page_url LIKE '{url_pat2}'){fts_where_sql}", tuple(params[1:])) total = c.fetchone()[0] if total > 0: # 三步: ①只取 rowid (idx_gov_raw_page_url 覆盖索引 1.5s) # ②批量 IN 查 publish_date (主键 0.1s) # ③Python 排序分页 + 主键取数 (0.0s) # 直接 SELECT 列 / 子查询 ORDER BY 都会逐行回表 ~20s c.execute(f"SELECT r.rowid FROM gov_raw r WHERE (r.page_url LIKE '{url_pat}' OR r.page_url LIKE '{url_pat2}'){fts_where_sql}", tuple(params[1:])) ids = [r[0] for r in c.fetchall()] if not ids: conn.close() return [], 0, False ph_ids = ','.join('?' * len(ids)) c.execute(f'SELECT id, publish_date FROM gov_raw r WHERE r.id IN ({ph_ids})', ids) id_dates = c.fetchall() # 按日期倒序 (空日期排最后), 稳定分页 id_dates.sort(key=lambda x: x[1] or '', reverse=True) page_ids = [rid for rid, _ in id_dates[offset:offset + fetch_size]] has_more = len(page_ids) > page_size if has_more: page_ids = page_ids[:page_size] if not page_ids: rows = [] else: ph2 = ','.join('?' * len(page_ids)) c.execute(f'SELECT * FROM gov_raw r WHERE r.id IN ({ph2})', page_ids) rows = [dict(r) for r in c.fetchall()] conn.close() return rows, total, has_more else: # page_url 无命中 → 单英文词(域名/日期/编号)直接返回空， # 避免落 FTS 表 LIKE 全扫卡死 (如 '2026-08-24' 0 命中时) conn.close() return [], 0, False except Exception: pass # page_url 无命中时兜底 title/site_name LIKE (可能慢但仅限异常) # 中文 / 多词 / 排除词 / 单英文词(域名URL) → LIKE 逐词匹配 if contains_chinese(q) or exc_terms or len(inc_terms) > 1 or (len(inc_terms) == 1 and not contains_chinese(q)): # 纯中文多词(每词≥3字, 空格=AND) → trigram FTS 优先; 排除词用 NOT LIKE 二次过滤 # 0命中不直接空返 → 继续落 bigram/轻量LIKE 兜底 (bigram 2026-09-03 已恢复) if contains_chinese(q) and all(len(t) >= 3 for t in inc_terms + exc_terms): try: _fts_q = sanitize_fts_query(' '.join(inc_terms)) if inc_terms else '' _w_list = ['gov_search MATCH ?'] _fparams = [_fts_q] if where_parts: _w_list.extend(where_parts) _fparams.extend(params[1:]) for _t in exc_terms: _w_list.append('(r.title NOT LIKE ? AND r.site_name NOT LIKE ? AND r.page_url NOT LIKE ?)') _fparams.extend(['%' + _t + '%'] * 3) _w_all = ' AND '.join(_w_list) c.execute(f'SELECT COUNT(*) FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE {_w_all}', _fparams) total = c.fetchone()[0] if total > 0: c.execute(f'SELECT r.*, rank FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE {_w_all} ORDER BY {order_sql} LIMIT ? OFFSET ?', _fparams + [fetch_size, offset]) rows = [dict(r) for r in c.fetchall()] has_more = len(rows) > page_size if has_more: rows = rows[:page_size] conn.close() return rows, total, has_more except sqlite3.OperationalError: pass like_conds = [] like_params = [] for t in inc_terms: like_conds.append('(r.title LIKE ? OR r.site_name LIKE ? OR r.page_url LIKE ?)') like_params.extend(['%' + t + '%', '%' + t + '%', '%' + t + '%']) for t in exc_terms: like_conds.append('(r.title NOT LIKE ? AND r.site_name NOT LIKE ? AND r.page_url NOT LIKE ?)') like_params.extend(['%' + t + '%', '%' + t + '%', '%' + t + '%']) if not like_conds: rows = []; total = 0; has_more = False conn.close() return rows, total, has_more # 中文词(≥2字) → bigram 索引快路径 (2026-09-03 恢复; LIKE %xx% 全扫 45s, gov_bigram 索引 0.01s) # 覆盖: 2字词直接 gram; ≥3字词拆 2-gram 序列交集 (FTS trigram 对错序词 0 命中时不再落 LIKE 全扫) # 含标点 token (项目-招标 / 环境:保护) 用 token_grams 提取中文段 gram, 不再触发 FTS 语法错误 if (inc_terms and all(len(t) >= 2 and contains_chinese(t) and token_grams(t) for t in inc_terms)): try: # 交集 rowid 集合 (纯 bigram 索引 COUNT, 不回表 ~0.01s) # ≥3字词: 拆 2-gram 序列 (醋酸乙烯 -> 醋酸/酸乙/乙烯), 各 gram 交集 ≈ 包含该词 id_sets = [] for t in inc_terms: grams = token_grams(t) term_ids = None for g in grams: c.execute('SELECT rowid FROM gov_bigram WHERE gram = ?', (g,)) s = set(r[0] for r in c.fetchall()) if term_ids is None: term_ids = s else: term_ids &= s if not term_ids: break id_sets.append(term_ids or set()) ids = set.intersection(*id_sets) if id_sets else set() if not ids: # 交集为空 → 直接返回空 (不再落 LIKE 全扫 45s) conn.close() return [], 0, False if ids: id_list = list(ids) # 排除词 → bigram 集合差 (title/site_name 不含该词 ≈ gram 集合差) # NOT LIKE 全表扫 37s / EXISTS COUNT 37s / 分批 IN 26s 全卡死； # bigram 集合差 0.03s。page_url 排除差异极少见，可接受。 for t in exc_terms: exc_grams = token_grams(t) if not exc_grams: continue for g in exc_grams: c.execute('SELECT rowid FROM gov_bigram WHERE gram = ?', (g,)) exc_s = set(r[0] for r in c.fetchall()) ids = ids - exc_s id_list = list(ids) if not id_list: conn.close() return [], 0, False # 带日期/行业过滤时需回表精确计数; 否则用 bigram 集合大小 extra_join = ' AND ' + ' AND '.join(where_parts) if where_parts else '' if where_parts: # 大集合(如"项目"26万)时禁止 WHERE id IN (N个参数) —— SQLite 999 变量上限会抛异常 # 被 except 吞掉后落 FTS 表 LIKE 全扫卡死。改用 EXISTS + 过滤条件直接 COUNT (流式, 走日期索引) gram_counts = [len(token_grams(t)) for t in inc_terms] ex_conds = ' AND '.join( ' AND '.join(f'EXISTS (SELECT 1 FROM gov_bigram b{i}_{j} WHERE b{i}_{j}.rowid=r.id AND b{i}_{j}.gram=?)' for j in range(gram_counts[i])) for i in range(len(inc_terms))) ex_params = [] for t in inc_terms: ex_params.extend(token_grams(t)) count_sql = f'SELECT COUNT(*) FROM gov_raw r WHERE {ex_conds}{extra_join}' c.execute(count_sql, ex_params) total = c.fetchone()[0] else: total = len(id_list) if total == 0: conn.close() return [], 0, False # 分页: 从 gov_raw 按 publish_date 索引倒序扫描 + EXISTS 校验 (0.00s, 找到即停) # 避免大词(公示26万)全集合排序; 日期索引保证流式取到当页即止 # ≥3字词同样拆 2-gram, 每个 gram 都须 EXISTS (AND 语义) gram_counts = [len(token_grams(t)) for t in inc_terms] ex_conds = ' AND '.join( ' AND '.join(f'EXISTS (SELECT 1 FROM gov_bigram b{i}_{j} WHERE b{i}_{j}.rowid=r.id AND b{i}_{j}.gram=?)' for j in range(gram_counts[i])) for i in range(len(inc_terms))) ex_params = [] for t in inc_terms: ex_params.extend(token_grams(t)) if where_parts: where_sql_join = extra_join else: where_sql_join = '' # 排除词 → 分页同样排除 (多取3倍再按剩余集合过滤, 保证页满; 排除比例高的极端场景分页会略少, 但绝不卡死) if exc_terms and id_list: limit_sql = f'SELECT r.id FROM gov_raw r WHERE {ex_conds}{where_sql_join} ORDER BY r.publish_date DESC LIMIT ? OFFSET ?' c.execute(limit_sql, ex_params + [fetch_size * 3, offset]) page_ids = [r[0] for r in c.fetchall()] page_ids = [pid for pid in page_ids if pid in ids] else: limit_sql = f'SELECT r.id FROM gov_raw r WHERE {ex_conds}{where_sql_join} ORDER BY r.publish_date DESC LIMIT ? OFFSET ?' c.execute(limit_sql, ex_params + [fetch_size, offset]) page_ids = [r[0] for r in c.fetchall()] has_more = len(page_ids) > page_size if has_more: page_ids = page_ids[:page_size] if page_ids: ph2 = ','.join('?' * len(page_ids)) c.execute(f'SELECT * FROM gov_raw r WHERE r.id IN ({ph2})', page_ids) rows = [dict(r) for r in c.fetchall()] else: rows = [] conn.close() return rows, total, has_more except Exception: pass # Use FTS table for LIKE (fast, ~382K rows) JOIN gov_raw (rowid 已对齐) fts_conds = [] for t in inc_terms: fts_conds.append('(s.title LIKE ? OR s.site_name LIKE ? OR r.page_url LIKE ?)') for t in exc_terms: fts_conds.append('(s.title NOT LIKE ? AND s.site_name NOT LIKE ? AND r.page_url NOT LIKE ?)') fts_where = ' AND '.join(fts_conds) fts_params = list(like_params) + params[1:] if where_parts: raw_where = ' AND '.join(where_parts) fts_where += ' AND ' + raw_where c.execute(f'SELECT COUNT(*) FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE {fts_where}', fts_params) total = c.fetchone()[0] c.execute(f'SELECT r.*, "" as rank FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE {fts_where} ORDER BY r.publish_date DESC LIMIT ? OFFSET ?', fts_params + [fetch_size, offset]) rows = [dict(r) for r in c.fetchall()] has_more = (offset + len(rows)) < total if has_more: rows = rows[:page_size] conn.close() return rows, total, has_more else: fts_where_sql = '' if where_parts: fts_where_sql = ' AND ' + ' AND '.join(where_parts) try: c.execute(f'SELECT COUNT(*) FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE gov_search MATCH ?{fts_where_sql}', (sanitize_fts_query(q),) + tuple(params[1:])) total = c.fetchone()[0] c.execute(f'SELECT r.*, rank FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE gov_search MATCH ?{fts_where_sql} ORDER BY {order_sql} LIMIT ? OFFSET ?', (sanitize_fts_query(q),) + tuple(params[1:]) + (fetch_size, offset)) except sqlite3.OperationalError: like_q = '%' + q + '%' c.execute(f'SELECT COUNT(*) FROM gov_raw WHERE (title LIKE ? OR summary LIKE ?){(" AND " + " AND ".join(where_parts)) if where_parts else ""}', (like_q, like_q) + tuple(params[1:])) total = c.fetchone()[0] c.execute(f'SELECT r.*, "" as rank FROM gov_raw r WHERE (title LIKE ? OR summary LIKE ?){(" AND " + " AND ".join(where_parts)) if where_parts else ""} ORDER BY r.publish_date DESC LIMIT ? OFFSET ?', (like_q, like_q) + tuple(params[1:]) + (fetch_size, offset)) else: where = 'WHERE ' + ' AND '.join(where_parts) if where_parts else '' c.execute(f'SELECT COUNT(*) FROM gov_raw r {where}', params) total = c.fetchone()[0] c.execute(f'SELECT r.* FROM gov_raw r {where} ORDER BY {order_sql} LIMIT ? OFFSET ?', params + [fetch_size, offset]) rows = [dict(r) for r in c.fetchall()] has_more = len(rows) > page_size if has_more: rows = rows[:page_size] conn.close() return rows, total, has_more def get_detail(record_id): conn = get_db(); c = conn.cursor() c.execute("SELECT * FROM gov_raw WHERE id=?", (record_id,)) r = c.fetchone() conn.close() return dict(r) if r else None # ─── CCEUP/中项网 DB ─── def cceup_search(q, page=1, per=20, daterange='all', industry=None): if not os.path.exists(CONTACT_DB_PATH): return 0, [] db = sqlite3.connect(CONTACT_DB_PATH); db.row_factory = sqlite3.Row conds, params = [], [] if q: inc, exc = parse_google_query(q) for t in inc: conds.append('(project_id LIKE ? OR project_name LIKE ? OR owner_company LIKE ? OR industry LIKE ? OR province LIKE ? OR city LIKE ?)') params.extend(['%'+t+'%']*6) for t in exc: conds.append('(project_id NOT LIKE ? AND project_name NOT LIKE ? AND owner_company NOT LIKE ? AND industry NOT LIKE ? AND province NOT LIKE ? AND city NOT LIKE ?)') params.extend(['%'+t+'%']*6) if industry and industry != 'all' and industry in INDUSTRY_MAP: kws = INDUSTRY_MAP[industry]['kw'][:4] if kws: sub_conds = ' OR '.join(['industry LIKE ?'] * len(kws)) conds.append('(' + sub_conds + ')') params.extend(['%'+k+'%' for k in kws]) if daterange and daterange != 'all': days = {'1d':1,'7d':7,'30d':30}[daterange] conds.append(f"(publish_date GLOB '20[0-9][0-9]-[0-9][0-9]-[0-9][0-9]*' AND SUBSTR(publish_date,1,10) >= date('now', '-{days} days'))") wh = 'WHERE '+' AND '.join(conds) if conds else '' total = db.execute('SELECT COUNT(*) FROM cceup_projects '+wh, params).fetchone()[0] rows = [dict(r) for r in db.execute('SELECT * FROM cceup_projects '+wh+' ORDER BY publish_date DESC, project_id LIMIT ? OFFSET ?', params+[per,(page-1)*per])] db.close() return total, rows def cceup_contacts_count(pid): if not os.path.exists(CONTACT_DB_PATH): return 0 db = sqlite3.connect(CONTACT_DB_PATH) total = db.execute('SELECT COUNT(*) FROM cceup_contacts WHERE project_id=?', (pid,)).fetchone()[0] db.close() return total # ─── ZNLH/中能联合 DB ─── def znlh_search(q, page=1, per=20, daterange='all', industry=None): if not os.path.exists(ZNLH_DB_PATH): return 0, [] db = sqlite3.connect(ZNLH_DB_PATH); db.row_factory = sqlite3.Row conds, params = [], [] if q: inc, exc = parse_google_query(q) for t in inc: conds.append('(project_id LIKE ? OR project_name LIKE ? OR owner_company LIKE ? OR industry LIKE ? OR province LIKE ? OR city LIKE ?)') params.extend(['%'+t+'%']*6) for t in exc: conds.append('(project_id NOT LIKE ? AND project_name NOT LIKE ? AND owner_company NOT LIKE ? AND industry NOT LIKE ? AND province NOT LIKE ? AND city NOT LIKE ?)') params.extend(['%'+t+'%']*6) if industry and industry != 'all' and industry in INDUSTRY_MAP: kws = INDUSTRY_MAP[industry]['kw'][:4] if kws: sub_conds = ' OR '.join(['industry LIKE ?'] * len(kws)) conds.append('(' + sub_conds + ')') params.extend(['%'+k+'%' for k in kws]) if daterange and daterange != 'all': days = {'1d':1,'7d':7,'30d':30}[daterange] conds.append(f"(publish_date GLOB '20[0-9][0-9]-[0-9][0-9]-[0-9][0-9]*' AND SUBSTR(publish_date,1,10) >= date('now', '-{days} days'))") wh = 'WHERE '+' AND '.join(conds) if conds else '' total = db.execute('SELECT COUNT(*) FROM znlh_projects '+wh, params).fetchone()[0] rows = [dict(r) for r in db.execute('SELECT * FROM znlh_projects '+wh+' ORDER BY publish_date DESC, project_id LIMIT ? OFFSET ?', params+[per,(page-1)*per])] db.close() return total, rows def znlh_detail(pid): if not os.path.exists(ZNLH_DB_PATH): return None db = sqlite3.connect(ZNLH_DB_PATH); db.row_factory = sqlite3.Row r = db.execute('SELECT * FROM znlh_projects WHERE project_id=?',(pid,)).fetchone() if not r: db.close(); return None p = dict(r) p['contacts'] = [dict(r) for r in db.execute('SELECT * FROM znlh_contacts WHERE project_id=? ORDER BY id',(pid,))] db.close(); return p # ─── ZC/中策大数据 DB ─── def zc_search(q, page=1, per=20, daterange='all', industry=None): if not os.path.exists(ZC_DB_PATH): return 0, [] db = sqlite3.connect(ZC_DB_PATH); db.row_factory = sqlite3.Row conds, params = [], [] if q: inc, exc = parse_google_query(q) for t in inc: conds.append('(project_id LIKE ? OR project_name LIKE ? OR industry LIKE ? OR province LIKE ? OR city LIKE ?)') params.extend(['%'+t+'%']*5) for t in exc: conds.append('(project_id NOT LIKE ? AND project_name NOT LIKE ? AND industry NOT LIKE ? AND province NOT LIKE ? AND city NOT LIKE ?)') params.extend(['%'+t+'%']*5) if industry and industry != 'all' and industry in INDUSTRY_MAP: kws = INDUSTRY_MAP[industry]['kw'][:4] if kws: sub_conds = ' OR '.join(['industry LIKE ?'] * len(kws)) conds.append('(' + sub_conds + ')') params.extend(['%'+k+'%' for k in kws]) if daterange and daterange != 'all': days = {'1d':1,'7d':7,'30d':30}[daterange] conds.append(f"(publish_date GLOB '20[0-9][0-9]-[0-9][0-9]-[0-9][0-9]*' AND SUBSTR(publish_date,1,10) >= date('now', '-{days} days'))") wh = 'WHERE '+' AND '.join(conds) if conds else '' total = db.execute('SELECT COUNT(*) FROM zc_projects '+wh, params).fetchone()[0] rows = [dict(r) for r in db.execute('SELECT * FROM zc_projects '+wh+' ORDER BY publish_date DESC, project_id LIMIT ? OFFSET ?', params+[per,(page-1)*per])] db.close() return total, rows def zc_detail(pid): if not os.path.exists(ZC_DB_PATH): return None db = sqlite3.connect(ZC_DB_PATH); db.row_factory = sqlite3.Row r = db.execute('SELECT * FROM zc_projects WHERE project_id=?',(pid,)).fetchone() if not r: db.close(); return None p = dict(r) p['contacts'] = [dict(r) for r in db.execute('SELECT * FROM zc_contacts WHERE project_id=? ORDER BY id',(pid,))] db.close(); return p # ══════════════════════════════════════════════ # HTTP Handler # ══════════════════════════════════════════════ # ─── Crawler DB ─── def crawler_detail(pid): if not os.path.exists(DB_PATH): return None db = sqlite3.connect(DB_PATH); db.row_factory = sqlite3.Row try: r = db.execute('SELECT * FROM gov_raw WHERE id=?',(pid,)).fetchone() if not r: return None d = dict(r) d['url'] = d.get('page_url') or d.get('source_url') or '' d['domain'] = d.get('site_name') or '' return d finally: db.close() # ─── Project visit tracking helpers ─── def _track_project_visit(ptype, pid, username): if not username or not pid: return db = sqlite3.connect(SESSION_DB) try: db.execute('INSERT INTO project_visits (project_type, project_id, username) VALUES (?,?,?)', (ptype, str(pid), username)) db.commit() except: pass finally: db.close() def _get_project_visitors(ptype, pid, limit=3): if not pid: return 0, [] db = sqlite3.connect(SESSION_DB) db.row_factory = sqlite3.Row try: total = db.execute('SELECT COUNT(*) FROM project_visits WHERE project_type=? AND project_id=?', (ptype, str(pid))).fetchone()[0] recent = [dict(r) for r in db.execute( 'SELECT DISTINCT username, visited_at FROM project_visits WHERE project_type=? AND project_id=? ORDER BY visited_at DESC LIMIT ?', (ptype, str(pid), limit))] unique_count = db.execute('SELECT COUNT(DISTINCT username) FROM project_visits WHERE project_type=? AND project_id=?', (ptype, str(pid))).fetchone()[0] return unique_count, recent except: return 0, [] finally: db.close() # ─── Login / Session ─── ALLOWED_USERS = {'wjianning', 'lzhong', 'steveou', 'schengbo', 'admin'} SESSION_DB = '/root/contact.db' def init_session_db(): db = sqlite3.connect(SESSION_DB) db.execute('''CREATE TABLE IF NOT EXISTS sessions ( token TEXT PRIMARY KEY, username TEXT NOT NULL, created_at TEXT DEFAULT (datetime('now','localtime')), last_seen TEXT DEFAULT (datetime('now','localtime')) )''') db.execute('''CREATE TABLE IF NOT EXISTS usage_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL, action TEXT NOT NULL, page TEXT DEFAULT '', query TEXT DEFAULT '', ip TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now','localtime')) )''') db.execute('''CREATE TABLE IF NOT EXISTS project_visits ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_type TEXT NOT NULL, project_id TEXT NOT NULL, username TEXT NOT NULL, visited_at TEXT DEFAULT (datetime('now','localtime')) )''') db.execute('''CREATE INDEX IF NOT EXISTS idx_pv_lookup ON project_visits(project_type, project_id, visited_at DESC)''') db.commit(); db.close() init_session_db() ensure_industry_column() # ─── HTTP Handler ─── class SearchHandler(http.server.BaseHTTPRequestHandler): def _proxy_tampermonkey(self, raw_path): """反向代理 /tm/* → 127.0.0.1:8010 (Tampermonkey script center)""" import http.client # 去掉 /tm 前缀，转发到 8010 target = raw_path if target.startswith('/tm'): target = target[len('/tm'):] or '/' try: conn = http.client.HTTPConnection('127.0.0.1', 8010, timeout=15) conn.request('GET', target, headers={'X-Forwarded-Prefix': '/tm'}) resp = conn.getresponse() body = resp.read() self.send_response(resp.status) for k, v in resp.getheaders(): if k.lower() not in ('content-length', 'transfer-encoding', 'connection'): self.send_header(k, v) self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) conn.close() except Exception as e: try: self.send_response(502) self.send_header('Content-Type', 'text/plain; charset=utf-8') self.send_header('Content-Length', str(len(str(e)) + 40)) self.end_headers() self.wfile.write(('Tampermonkey 服务不可用: %s' % e).encode('utf-8')) except Exception: pass def _send_short_query_notice(self, path, kw): # 2026-09-03: 关键词 ≤2 字符守卫 (gov_bigram 已下线, trigram FTS 需 ≥3 字符) _body = ('提示' '' '
' '

关键词过短

' '

关键词 “' + html.escape(kw) + '” 过短（≤2 个字符），' '无法进行有效检索。请至少输入 3 个字符，建议使用完整的公司名称或更具体的项目关键词。

' '

← 返回搜索

' '
') self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(_body.encode('utf-8')) def do_GET(self): parsed = urllib.parse.urlparse(self.path) params = urllib.parse.parse_qs(parsed.query) # Session check self.username = self._get_session_user() if parsed.path.startswith('/crawler/admin'): self.username = self.username or 'admin' # bypass login for admin page # Tampermonkey script center proxy (/tm/* → 127.0.0.1:8010) if parsed.path.startswith('/tm/') or parsed.path == '/tm': self._proxy_tampermonkey(self.path) return if parsed.path in ('/login', '/logout'): if parsed.path == '/logout': self.send_response(302) self.send_header('Set-Cookie', 'session=; Path=/; Max-Age=0') self.send_header('Location', '/login') self.end_headers() return self.handle_login_page(params) return # Static files (no auth required) if parsed.path.startswith('/static/'): import os as _os, mimetypes _file_path = '/root/gov_crawler' + parsed.path if _os.path.isfile(_file_path): _content_type, _ = mimetypes.guess_type(_file_path) self.send_response(200) self.send_header('Content-Type', _content_type or 'application/octet-stream') self.send_header('Cache-Control', 'max-age=86400') self.end_headers() with open(_file_path, 'rb') as _f: self.wfile.write(_f.read()) else: self.send_response(404); self.end_headers(); self.wfile.write(b'404') return # Originals no-auth if parsed.path.startswith('/originals/detail'): self.handle_originals_detail(params) return if parsed.path.startswith('/originals'): self.handle_originals_page(params) return if not self.username: self.send_response(302) self.send_header('Location', '/login') self.end_headers() return self._log_usage(parsed.path, params) # 2026-09-03: 搜索词整体 ≤2 字符(去空格)守卫 — bigram 已恢复, 短词在组合内可由 bigram 命中; 仅整体过短不检索 if parsed.path in ('/', '/crawler', '/crawler/'): _kws = [params.get(_k, [''])[0].strip() for _k in ('q', 'q2', 'q3')] _kws = [_k for _k in _kws if _k] if _kws and len(''.join(_k.replace(' ', '') for _k in _kws)) <= 2: self._send_short_query_notice(parsed.path, _kws[0]) return # Track project visits on detail pages # ⚠️ /xxx/project/ 路径**不**在此跟踪——handle_db_detail 内部已 _track_project_visit（2506行）， # 这里再记一次会导致每次点击记录 2 次。 # 仅 /detail (handle_detail, 内部无跟踪) 在此记录。 _parsed_path = parsed.path _params = params if _parsed_path == '/detail' and 'id' in _params: _id = _params.get('id', ['0'])[0] _track_project_visit('gov', _id, getattr(self, 'username', 'anonymous')) # ZNLH routes if parsed.path in ('/znlh', '/znlh/'): self.handle_db_list('znlh', params) elif parsed.path.startswith('/znlh/project/'): self.handle_db_detail('znlh', parsed.path) # ZC routes elif parsed.path in ('/zc', '/zc/'): self.handle_db_list('zc', params) elif parsed.path.startswith('/zc/project/'): self.handle_db_detail('zc', parsed.path) # Crawler routes elif parsed.path in ('/crawler', '/crawler/'): self.handle_db_list('crawler', params) elif parsed.path.startswith('/crawler/project/'): self.handle_db_detail('crawler', parsed.path) # EIA routes elif parsed.path in ('/eia', '/eia/'): self.handle_db_list('eia', params) elif parsed.path.startswith('/eia/project/'): self.handle_db_detail('eia', parsed.path) # Original routes elif parsed.path == '/': self.handle_search_page(params) elif parsed.path == '/detail': self.handle_detail(params) elif parsed.path == '/cceup': self.handle_cceup_page(params) elif parsed.path == '/contact': self.handle_contact_page(params) elif parsed.path == '/ccpc': self.handle_cceup_projects(params) elif parsed.path == '/cceup/projects': self.handle_cceup_projects(params) # Temp DB route (crawler temporary storage, pre-sync) elif parsed.path in ('/crawler/temp', '/crawler/temp/'): self.handle_temp_db(params) elif parsed.path == '/crawler/temp/sync': self.handle_temp_sync() elif parsed.path.startswith('/cceup/project/') or parsed.path.startswith('/ccpc/project/'): self.handle_cceup_project_detail(self.path) elif parsed.path == '/dashboard': self.handle_dashboard() elif parsed.path == '/api/sources': self.handle_sources_api() elif parsed.path == '/api/visit': self.handle_visit(params) elif parsed.path == '/preview_md': self.send_preview_md() elif parsed.path == '/crawler/admin': self.send_response_html(200, render_admin_page()) elif parsed.path == '/crawler/admin/run-daily': import json result = run_daily_crawl() self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode()) elif parsed.path == '/crawler/admin/run-status': import json result = get_daily_run_status() self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode()) elif parsed.path == '/crawler/admin/set-industry': import json n = params.get('name', [''])[0] ind = params.get('industry', [''])[0] result = set_crawler_industry(n, ind) self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode()) elif parsed.path == '/crawler/admin/set-url': import json n = params.get('name', [''])[0] u = params.get('url', [''])[0] result = set_crawler_url(n, u) self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode()) elif parsed.path == '/crawler/admin/toggle-enabled': import json n = params.get('name', [''])[0] result = toggle_crawler_enabled(n) self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode()) elif parsed.path == '/crawler/admin/save-exclude': import json, os keywords = params.get('kw', [''])[0] try: with open(ADMIN_EXCLUDE_PATH, 'w') as f: f.write(keywords) result = {'success': True, 'keywords': keywords} except Exception as e: result = {'success': False, 'error': str(e)} self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode()) elif parsed.path == "/crawler/admin/analysis": import json from crawler_admin import get_analysis_data period = params.get("period", ["day"])[0] mode = params.get("mode", ["incremental"])[0] metric = params.get("metric", ["both"])[0] days = int(params.get("days", ["90"])[0]) result = get_analysis_data(period, mode, metric, days) self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode()) else: self.send_response(404); self.end_headers(); self.wfile.write(b"404") def handle_originals_page(self, params): q = self.get_param(params, 'q', '').strip() page = int(self.get_param(params, 'page', '1')) page_size = int(self.get_param(params, 'page_size', '20')) offset = (page - 1) * page_size ORIGINALS_DB = '/root/originals.db' if not os.path.exists(ORIGINALS_DB): self.send_response(200) self.send_header('Content-type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(('

Originals DB not found

').encode('utf-8')) return conn = sqlite3.connect(ORIGINALS_DB) c = conn.cursor() results = [] total = 0 if q: inc_terms, exc_terms = parse_google_query(q) conds = [] params = [] for t in inc_terms: conds.append('(title LIKE ? OR content LIKE ? OR id_code LIKE ?)') params.extend(['%' + t + '%'] * 3) for t in exc_terms: conds.append('(title NOT LIKE ? AND content NOT LIKE ? AND id_code NOT LIKE ?)') params.extend(['%' + t + '%'] * 3) where = ' AND '.join(conds) if conds else '1=0' total = c.execute('SELECT COUNT(*) FROM tsk_data WHERE ' + where, params).fetchone()[0] rows = c.execute('SELECT id, title, id_code, source, content FROM tsk_data WHERE ' + where + ' ORDER BY id LIMIT ? OFFSET ?', params + [page_size, offset]).fetchall() for r in rows: results.append({'id': r[0], 'title': r[1], 'id_code': r[2] or '', 'source': r[3], 'content': r[4] or ''}) # Get total count before closing total_count = 0 try: total_count = c.execute('SELECT COUNT(*) FROM tsk_data').fetchone()[0] except: pass conn.close() self.send_response(200) self.send_header('Content-type', 'text/html; charset=utf-8') self.end_headers() h = 'Originals_Data' h += '' h += '

🔍 原始数据库

' h += '

' + str(total_count) + ' records · Pharma / CPI / Plant

' h += '
' h += '' h += '
' if q: h += '

' + str(total) + ' 条结果

' if results: h += '
' for r in results: sc = r['source'] sl = {'pharma':'P Pharma','cpi':'C CPI','plant':'L Plant'}.get(sc, sc) sc_cls = {'pharma':'src-p','cpi':'src-c','plant':'src-l'}.get(sc, '') h += '
' h += '

' + html.escape(r['title']) + '

' h += '
' h += '' + (html.escape(r['id_code']) if r['id_code'] else '-') + '' h += '' + sl + '' h += '
' h += '
' tp = max(1, (total + page_size - 1) // page_size) if tp > 1: h += '
' for p in range(1, min(tp + 1, 51)): if p == page: h += '' + str(p) + ' ' else: h += '' + str(p) + ' ' if tp > 50: h += '... Last' h += '
' else: h += '

无结果

' h += '

"U+2190" 返回主页

' h += '' self.wfile.write(h.encode('utf-8')) def handle_originals_detail(self, params): record_id = int(self.get_param(params, 'id', '0')) ORIGINALS_DB = '/root/originals.db' if not os.path.exists(ORIGINALS_DB): self.send_response_html(404, 'Not found') return conn = sqlite3.connect(ORIGINALS_DB) row = conn.execute('SELECT id, title, id_code, source, content FROM tsk_data WHERE id=?', (record_id,)).fetchone() conn.close() if not row: self.send_response_html(404, 'Not found') return r = {'id': row[0], 'title': row[1], 'id_code': row[2] or '', 'source': row[3], 'content': row[4] or ''} sc = r['source'] sl = {'pharma':'P Pharma','cpi':'C CPI','plant':'L Plant'}.get(sc, sc) sc_cls = {'pharma':'src-p','cpi':'src-c','plant':'src-l'}.get(sc, '-') h = '' + html.escape(r['title'][:80]) + '' h += '' h += '← 返回列表' h += '
' h += '

' + html.escape(r['title']) + '

' h += '
' h += '' + (html.escape(r['id_code']) if r['id_code'] else '-') + '' h += '' + sl + '' h += 'ID: ' + str(r['id']) + '' h += '
' h += '
' + html.escape(re.sub(r'\n{3,}', '\n\n', r['content'] or '')) + '
' h += '
' self.send_response_html(200, h) def handle_temp_db(self, params): """Show temp_search.db contents with search.""" TEMP_DB = '/root/temp_search.db' TRACKING_FILE = '/root/.temp_sync_state.json' page = int(self.get_param(params, 'page', '1')) per = 20 offset = (page - 1) * per q = self.get_param(params, 'q', '').strip() sync_state = {} if os.path.exists(TRACKING_FILE): try: with open(TRACKING_FILE) as f: sync_state = json.load(f) except: pass rows = [] total = 0 temp_size = 0 last_synced = sync_state.get('last_synced_id', 0) error = None un_synced = 0 if os.path.exists(TEMP_DB): temp_size = os.path.getsize(TEMP_DB) try: db = sqlite3.connect(TEMP_DB) db.execute('PRAGMA busy_timeout=3000') c = db.cursor() if q: like = '%' + q + '%' total = c.execute( 'SELECT COUNT(*) FROM gov_raw WHERE title LIKE ? OR site_name LIKE ? OR page_url LIKE ?', (like, like, like) ).fetchone()[0] un_synced = c.execute( 'SELECT COUNT(*) FROM gov_raw WHERE id > ? AND (title LIKE ? OR site_name LIKE ? OR page_url LIKE ?)', (last_synced, like, like, like) ).fetchone()[0] c.execute( 'SELECT id, site_name, title, publish_date FROM gov_raw WHERE title LIKE ? OR site_name LIKE ? OR page_url LIKE ? ORDER BY id DESC LIMIT ? OFFSET ?', (like, like, like, per, offset) ) else: total = c.execute('SELECT COUNT(*) FROM gov_raw').fetchone()[0] un_synced = c.execute('SELECT COUNT(*) FROM gov_raw WHERE id > ?', (last_synced,)).fetchone()[0] c.execute( 'SELECT id, site_name, title, publish_date FROM gov_raw ORDER BY id DESC LIMIT ? OFFSET ?', (per, offset) ) rows = [{'id': r[0], 'site': r[1], 'title': r[2], 'date': r[3]} for r in c.fetchall()] db.close() except Exception as e: error = str(e) # Build HTML (string concat to avoid quoting issues) h = '' h += '' h += 'Temp DB - 临时数据' h += '
' h += '

🧊 Temp DB

' h += '← 主库
' if error: h += '
' + str(error) + '
' else: h += '
' label = '结果' if q else '总行数' h += '
' + str(total) + '
' + label + '
' h += '
' + str(un_synced) + '
待同步
' last_sv = sync_state.get('last_sync') or '—' h += '
' + last_sv + '
上次同步
' h += '
' + str(temp_size // 1024) + 'KB
文件大小
' h += '
' # Search bar qs = q.replace('"', '"') h += '
' h += '' h += '
' # Action buttons h += '
' h += '🔄 立即同步' clear_label = '🔄 刷新' if not q else '✖ 清除搜索' h += '' + clear_label + '' h += '
' if q: h += '
搜索 "' + q + '", 共 ' + str(total) + ' 条结果
' if rows: h += '' for r in rows: h += '' site = (r['site'] or '')[:80] title = (r['title'] or '')[:200] h += '' h += '' h += '' h += '
ID站点标题日期
' + str(r['id']) + '' + site + '' + title + '' + (r['date'] or '') + '
' total_pages = max(1, (total + per - 1) // per) if total_pages > 1: h += '
' for p in range(1, total_pages + 1): link = '/crawler/temp/?page=' + str(p) if q: link += '&q=' + q if p == page: h += '' + str(p) + '' else: h += '' + str(p) + '' h += '
' else: if q: h += '
未找到匹配结果
' else: h += '
暂无数据
' h += '
' h += 'temp_search.db | last_synced_id=' + str(last_synced) h += ' | 每日8:00自动同步
' h += '
' self.send_response_html(200, h) def handle_temp_sync(self): """Trigger immediate sync from temp to main.""" import subprocess try: r = subprocess.run(['python3', '/root/sync_temp_to_main.py'], capture_output=True, text=True, timeout=30) output = r.stdout or '' err = r.stderr or '' if err: output += f'\nSTDERR: {err}' if r.returncode != 0: output += f'\nExit code: {r.returncode}' except Exception as e: output = f'Error: {e}' self.send_response(200) self.send_header('Content-Type', 'text/plain; charset=utf-8') self.end_headers() self.wfile.write(output.encode()) def do_POST(self): length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode('utf-8', errors='replace') if length else '' params = urllib.parse.parse_qs(body) path = urllib.parse.urlparse(self.path).path if path == '/login': username = (params.get('username') or [''])[0].strip() if username in ALLOWED_USERS: token = str(uuid.uuid4()) db = sqlite3.connect(SESSION_DB) db.execute('INSERT OR REPLACE INTO sessions (token, username) VALUES (?,?)', (token, username)) db.commit(); db.close() self.send_response(302) self.send_header('Set-Cookie', f'session={token}; Path=/; Max-Age=2592000') self.send_header('Location', '/') self.end_headers() else: self.send_response(302) self.send_header('Location', '/login?error=1') self.end_headers() elif path == '/crawler/admin/run-daily': import json result = run_daily_crawl() self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode()) elif path == '/crawler/admin/run-status': import json result = get_daily_run_status() self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode()) else: self.send_response(404); self.end_headers(); self.wfile.write(b'404') def get_param(self, params, name, default=''): vals = params.get(name, [default]) return vals[0] if vals else default def _get_session_user(self): cookie = self.headers.get('Cookie', '') for part in cookie.split(';'): part = part.strip() if part.startswith('session='): token = part[8:] db = sqlite3.connect(SESSION_DB) try: row = db.execute('SELECT username FROM sessions WHERE token=?', (token,)).fetchone() if row: db.execute('UPDATE sessions SET last_seen=datetime("now","localtime") WHERE token=?', (token,)) db.commit() return row[0] if row else None finally: db.close() return None def _log_usage(self, path, params): if not self.username: return q = self.get_param(params, 'q', '') action = 'view' if q: action = 'search' elif path.startswith('/cceup/project/') or path.startswith('/crawler/project/') or path.startswith('/zc/project/') or path.startswith('/znlh/project/'): action = 'detail' db = sqlite3.connect(SESSION_DB) try: db.execute('INSERT INTO usage_log (username, action, page, query, ip) VALUES (?,?,?,?,?)', (self.username, action, path, q[:200], self.client_address[0])) db.commit() except: pass finally: db.close() def handle_login_page(self, params): error = 'error' in params h = '登录 - 项目信息检索' h += '' h += '

🔍 项目信息检索

请输入用户名登录

' if error: h += '
用户名无效，请重新输入
' h += '
' h += '
' h += '
数据无价 请勿外传
' self.send_response_html(200, h) def send_preview_md(self): try: with open('/root/preview_md.html', 'rb') as f: data = f.read() self.send_response(200) self.send_header('Content-Type','text/html; charset=utf-8') self.send_header('Content-Length',str(len(data))) self.end_headers() self.wfile.write(data) except Exception as e: self.send_response(500) self.end_headers() self.wfile.write(str(e).encode()) def send_response_html(self, code, html): b = html.encode() self.send_response(code) self.send_header('Content-Type','text/html; charset=utf-8') self.send_header('Content-Length',str(len(b))) self.end_headers() self.wfile.write(b) # ─── 通用数据库列表/详情 ─── def handle_db_list(self, db_type, params): page = int(self.get_param(params, 'page', '1')) q = self.get_param(params, 'q', '').strip() q2 = self.get_param(params, 'q2', '').strip() q3 = self.get_param(params, 'q3', '').strip() daterange = self.get_param(params, 'range', 'all') industry_filter = self.get_param(params, 'industry', '') per = 20 names = {'znlh':'中能联合','zc':'中策大数据','crawler':'EIA','eia':'EIA项目库'} title = names.get(db_type, db_type) total = 0 if db_type == 'crawler': has_more, rows = False, [] if os.path.exists(DB_PATH): db = get_db() c = db.cursor() fetch_size = per + 1 offset = (page-1)*per # 构建白名单关键词 SQL 条件（数据库层面过滤，确保 total 正确） # 语义：不再排除，而是"仅限给定关键词"（OR 匹配任一即保留） _include_kw_list = [] _include_cond_r = '' # 带 r. 前缀（FTS/LIKE JOIN 路径用） _include_cond_plain = '' # 无前缀（无词浏览路径用） _include_params = [] try: with open(ADMIN_EXCLUDE_PATH) as _ef: _include_kw = _ef.read().strip() if _include_kw: # 逗号或空格分隔（兼容中英文逗号） _include_list = [k.strip().lower() for k in re.split(r'[,，\s]+', _include_kw) if k.strip()] if _include_list: _include_kw_list = _include_list _include_cond_r = '(' + ' OR '.join( '(r.title LIKE ? OR r.site_name LIKE ? OR r.page_url LIKE ?)' for _ in _include_list) + ')' _include_cond_plain = '(' + ' OR '.join( '(title LIKE ? OR site_name LIKE ? OR page_url LIKE ?)' for _ in _include_list) + ')' _include_params = [f'%{_kw}%' for _kw in _include_list for _ in range(3)] except: pass # ── 2026-09-03 gov_entity 精确反查: 完整公司名/电话 (正文出现也命中) ── _ent_done = False if not q2 and not q3 and ' ' not in (q or '').strip(): _ep = None if q and not q.startswith('-'): _inc0, _exc0 = parse_google_query(q) if len(_inc0) == 1 and not _exc0: _t0 = _inc0[0] if contains_chinese(_t0) and len(_t0) >= 3: _ep = ('company', _t0) else: _dg = re.sub(r'\D', '', _t0) if 7 <= len(_dg) <= 13: _ep = ('phone', _dg) if _ep: try: _ec = ['e.etype = ?', 'e.value = ?'] _ev = [_ep[0], _ep[1]] if daterange and daterange != 'all': _days = {'1d': 1, '7d': 7, '30d': 30}[daterange] _ec.append("(r.publish_date GLOB '20[0-9][0-9]-[0-9][0-9]-[0-9][0-9]*' AND SUBSTR(r.publish_date,1,10) >= date('now', '-%d days'))" % _days) if industry_filter: _ec.append('r.industry = ?') _ev.append(industry_filter) if _include_kw_list: _ec.append('(' + ' OR '.join('(r.title LIKE ? OR r.site_name LIKE ? OR r.page_url LIKE ?)' for _ in _include_kw_list) + ')') _ev.extend([f'%{_k}%' for _k in _include_kw_list for _ in range(3)]) c.execute('SELECT COUNT(*) FROM gov_entity e JOIN gov_raw r ON r.id = e.doc_id WHERE ' + ' AND '.join(_ec), _ev) total = c.fetchone()[0] if total > 0: c.execute('SELECT r.* FROM gov_entity e JOIN gov_raw r ON r.id = e.doc_id WHERE ' + ' AND '.join(_ec) + ' ORDER BY r.publish_date DESC, r.id DESC LIMIT ? OFFSET ?', _ev + [fetch_size, offset]) _ent_rows = [dict(r) for r in c.fetchall()] rows = _ent_rows _ent_done = True except sqlite3.OperationalError: pass if q and not _ent_done: inc, exc = parse_google_query(q) fts_parts = [] for t in inc: fts_parts.append(t) for t in exc: fts_parts.append('-' + t) fts_q = ' AND '.join(fts_parts) extra_parts = [] extra_params = [] if q2: extra_parts.append('(r.title LIKE ? OR r.source_url LIKE ? OR r.page_url LIKE ?)') extra_params.extend([f'%{q2}%']*3) if q3: extra_parts.append('(r.title LIKE ? OR r.source_url LIKE ? OR r.page_url LIKE ?)') extra_params.extend([f'%{q3}%']*3) if daterange and daterange != 'all': days = {'1d':1,'7d':7,'30d':30}[daterange] extra_parts.append(f"(r.publish_date GLOB '20[0-9][0-9]-[0-9][0-9]-[0-9][0-9]*' AND SUBSTR(r.publish_date,1,10) >= date('now', '-{days} days'))") if industry_filter: extra_parts.append('(r.industry = ?)') extra_params.append(industry_filter) # 白名单关键词**不**加入 extra_parts —— bigram 快路径用集合差、FTS/LIKE 路径单独拼接。 # (避免 bigram COUNT/SELECT 里带白名单 LIKE %kw% 全表扫卡死) extra_sql = ' AND ' + ' AND '.join(extra_parts) if extra_parts else '' _include_sql_r = (' AND ' + _include_cond_r) if _include_cond_r else '' _include_all_params = list(_include_params) # ALL queries try FTS first (Chinese included) # 中文单词>=3字符：先用 trigram FTS（比 LIKE 快100倍） # 日期/行业过滤(extra_parts)直接拼进 FTS MATCH——不要 LIMIT 5000 截断(会丢新入库的高rowid记录) if contains_chinese(q): _fts_ok = False if not exc and all(len(t) >= 3 for t in inc) and len(q.strip()) >= 3: try: _fts_q = sanitize_fts_query(q) c.execute(f"SELECT COUNT(*) FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE gov_search MATCH ?{extra_sql}{_include_sql_r}", (_fts_q, *extra_params, *_include_all_params)) total = c.fetchone()[0] c.execute(f"SELECT r.* FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE gov_search MATCH ?{extra_sql}{_include_sql_r} ORDER BY r.publish_date DESC, r.id DESC LIMIT ? OFFSET ?", (_fts_q, *extra_params, *_include_all_params, fetch_size, offset)) _fts_ok = True except: pass if not _fts_ok: # 先用FTS缩小范围，再在内存中过滤 try: _fts_rowids = set() for _t in inc: if len(_t) >= 3: c.execute("SELECT rowid FROM gov_search WHERE gov_search MATCH ? LIMIT 5000", (sanitize_fts_query(_t),)) _ids = set(r[0] for r in c.fetchall()) if _fts_rowids: _fts_rowids = _fts_rowids & _ids else: _fts_rowids = _ids if _fts_rowids: _id_list = ",".join(str(i) for i in list(_fts_rowids)[:5000]) _where_extra = " AND r.id IN (" + _id_list + ")" else: _where_extra = "" # exclusion handled via FTS NOT terms in query except: _where_extra = "" # LIKE 条件（在FTS缩小后的ID范围内） like_conds = [] like_params = [] for t in inc: like_conds.append('(r.title LIKE ? OR r.site_name LIKE ? OR r.summary LIKE ?)') like_params.extend(['%' + t + '%', '%' + t + '%', '%' + t + '%']) for t in exc: like_conds.append('(r.title NOT LIKE ? AND r.site_name NOT LIKE ? AND r.summary NOT LIKE ?)') like_params.extend(['%' + t + '%', '%' + t + '%', '%' + t + '%']) like_where = ' AND '.join(like_conds) if extra_parts: like_where += ' AND ' + ' AND '.join(extra_parts) like_params += extra_params if _include_cond_r: like_where += _include_sql_r like_params += _include_all_params if _where_extra: like_where += _where_extra # bigram 快路径: inc 全是中文≥2字且有有效 gram 时, 用 gov_bigram 交集缩小范围 # 避免 LIKE %xx% 全表扫 + 白名单 LIKE 全表扫双重慢(45s+卡死) # (2字词"环境"不满足 FTS trigram, 会走到这里) _bigram_ok = False # 2026-09-03: gov_bigram 已恢复 → 该快路径重新启用 if (inc and not _where_extra and all(len(t) >= 2 and contains_chinese(t) and token_grams(t) for t in inc)): try: _b_sets = [] for _t in inc: _g_s = None for _g in token_grams(_t): c.execute('SELECT rowid FROM gov_bigram WHERE gram = ?', (_g,)) _s = set(r[0] for r in c.fetchall()) _g_s = _s if _g_s is None else (_g_s & _s) if not _g_s: break _b_sets.append(_g_s or set()) _b_ids = set.intersection(*_b_sets) if _b_sets else set() if _b_ids: # 白名单过滤 (bigram 只索引 title/site_name; page_url 白名单差异极少见) # OR 语义: 任一关键词命中即保留 → 各关键词集合取并集再交集 if _include_kw_list: _kw_union = set() for _kw in _include_kw_list: _kw_grams = token_grams(_kw) or ([_kw] if len(_kw) == 2 else []) _kw_s = None for _g in _kw_grams: c.execute('SELECT rowid FROM gov_bigram WHERE gram = ?', (_g,)) _s = set(r[0] for r in c.fetchall()) _kw_s = _s if _kw_s is None else (_kw_s & _s) if _kw_s is None: _kw_s = set() _kw_union |= _kw_s _b_ids = _b_ids & _kw_union # 排除词 → bigram 集合差 for _t in exc: for _g in token_grams(_t): c.execute('SELECT rowid FROM gov_bigram WHERE gram = ?', (_g,)) _exc_s = set(r[0] for r in c.fetchall()) _b_ids = _b_ids - _exc_s if _b_ids: # 大集合时禁止 IN 超参数; 用 EXISTS + 过滤条件流式计数 # inc 的 gram 条件 _b_gram_counts = [len(token_grams(t)) for t in inc] _b_ex_conds = ' AND '.join( ' AND '.join(f'EXISTS (SELECT 1 FROM gov_bigram b{i}_{j} WHERE b{i}_{j}.rowid=r.id AND b{i}_{j}.gram=?)' for j in range(_b_gram_counts[i])) for i in range(len(inc))) _b_ex_params = [] for _t in inc: _b_ex_params.extend(token_grams(_t)) # 白名单 → OR of (AND of EXISTS gram per kw) —— 任一关键词命中即保留 _b_wl_conds = [] for _ki, _kw in enumerate(_include_kw_list): _b_kw_grams = token_grams(_kw) or ([_kw] if len(_kw) == 2 else []) if _b_kw_grams: _b_wl_conds.append('(' + ' AND '.join( f'EXISTS (SELECT 1 FROM gov_bigram w{_ki}_{j} WHERE w{_ki}_{j}.rowid=r.id AND w{_ki}_{j}.gram=?)' for j in range(len(_b_kw_grams))) + ')') _b_ex_params.extend(_b_kw_grams) # 排除词 → NOT EXISTS (含该词则排除) for _t in exc: for _g in (token_grams(_t) or ([_t] if len(_t) == 2 else [])): _b_wl_conds.append('NOT EXISTS (SELECT 1 FROM gov_bigram e0 WHERE e0.rowid=r.id AND e0.gram=?)') _b_ex_params.append(_g) if _b_wl_conds: _b_wl_sql = ' AND (' + ' OR '.join(_b_wl_conds) + ')' else: _b_wl_sql = '' # extra_parts (日期/行业) 直接拼入 if extra_parts: _b_extra_join = ' AND ' + ' AND '.join(extra_parts) else: _b_extra_join = '' _b_count_sql = f'SELECT COUNT(*) FROM gov_raw r WHERE {_b_ex_conds}{_b_wl_sql}{_b_extra_join}' c.execute(_b_count_sql, _b_ex_params + extra_params if extra_parts else _b_ex_params) total = c.fetchone()[0] if total > 0: _b_limit_sql = f'SELECT r.id FROM gov_raw r WHERE {_b_ex_conds}{_b_wl_sql}{_b_extra_join} ORDER BY r.publish_date DESC, r.id DESC LIMIT ? OFFSET ?' c.execute(_b_limit_sql, (_b_ex_params + extra_params if extra_parts else _b_ex_params) + [fetch_size, offset]) _b_page_ids = [r[0] for r in c.fetchall()] has_more = len(_b_page_ids) > per if has_more: _b_page_ids = _b_page_ids[:per] if _b_page_ids: _b_ph = ','.join('?' * len(_b_page_ids)) c.execute(f'SELECT r.* FROM gov_raw r WHERE r.id IN ({_b_ph})', _b_page_ids) rows = [dict(r) for r in c.fetchall()] else: rows = [] _bigram_ok = True except Exception: _bigram_ok = False if not _bigram_ok: c.execute(f'SELECT COUNT(*) FROM gov_raw r WHERE {like_where}', like_params) total = c.fetchone()[0] c.execute(f'SELECT r.*, "" as rank FROM gov_raw r WHERE {like_where} ORDER BY r.publish_date DESC, r.id DESC LIMIT ? OFFSET ?', like_params + [fetch_size, offset]) else: try: _fts_q = sanitize_fts_query(fts_q) c.execute(f'SELECT COUNT(*) FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE gov_search MATCH ?{extra_sql}', (_fts_q, *extra_params)) total = c.fetchone()[0] c.execute(f'SELECT r.* FROM gov_search s JOIN gov_raw r ON s.rowid=r.id WHERE gov_search MATCH ?{extra_sql} ORDER BY r.publish_date DESC, r.id DESC LIMIT ? OFFSET ?', (_fts_q, *extra_params, fetch_size, offset)) except sqlite3.OperationalError: # FTS语法错误，降级到LIKE like_q = '%' + q + '%' c.execute('SELECT COUNT(*) FROM gov_raw r WHERE (r.title LIKE ? OR r.summary LIKE ?) ORDER BY r.publish_date DESC, r.id DESC LIMIT ? OFFSET ?', (like_q, like_q, fetch_size, offset)) total = 0 rows = [dict(r) for r in c.fetchall()] # 无法直接用FTS加速的fallback，白名单过滤（仅限给定关键词） if _include_kw_list: rows = [r for r in rows if any(kw in (r.get('title','') or '').lower() or kw in (r.get('site_name','') or '').lower() or kw in (r.get('page_url','') or '').lower() for kw in _include_kw_list)] elif not q: # 无搜索词：按日期浏览 where_parts = [] sql_params = [] if q2: where_parts.append('(title LIKE ? OR source_url LIKE ? OR page_url LIKE ?)') sql_params.extend([f'%{q2}%']*3) if q3: where_parts.append('(title LIKE ? OR source_url LIKE ? OR page_url LIKE ?)') sql_params.extend([f'%{q3}%']*3) if daterange and daterange != 'all': days = {'1d':1,'7d':7,'30d':30}[daterange] where_parts.append(f"(publish_date GLOB '20[0-9][0-9]-[0-9][0-9]-[0-9][0-9]*' AND SUBSTR(publish_date,1,10) >= date('now', '-{days} days'))") if industry_filter: where_parts.append('industry = ?') sql_params.append(industry_filter) # 白名单关键词（无词浏览路径，无 r. 前缀） if _include_cond_plain: where_parts.append(_include_cond_plain) sql_params.extend(_include_params) where_sql = ' WHERE ' + ' AND '.join(where_parts) if where_parts else '' # 无词浏览 + 白名单: LIKE %kw% 无法用索引, 全表 COUNT 54s 卡死。 # 改用截断式计数: ORDER BY publish_date 走日期索引流式, 数到 5001 就停 (0.00s)。 # 超过 5000 条时 total 显示 5001(截断), admin 页展示"共 5001+ 条"语义。 if _include_kw_list: c.execute(f'SELECT COUNT(*) FROM (SELECT id FROM gov_raw{where_sql} ORDER BY publish_date DESC, id DESC LIMIT 5001)', sql_params) total = c.fetchone()[0] else: c.execute(f'SELECT COUNT(*) FROM gov_raw{where_sql}', sql_params) total = c.fetchone()[0] c.execute(f'SELECT * FROM gov_raw{where_sql} ORDER BY publish_date DESC, id DESC LIMIT ? OFFSET ?', sql_params + [fetch_size, offset]) rows = [dict(r) for r in c.fetchall()] # Post-fetch whitelist filter (applies to both query and no-query paths) # 白名单语义：仅保留匹配任一给定关键词的行（title/site_name/page_url） if _include_kw_list: rows = [r for r in rows if any( kw in (r.get('title','') or '').lower() or kw in (r.get('site_name','') or '').lower() or kw in (r.get('page_url','') or '').lower() for kw in _include_kw_list )] # Apply industry filter — 直接用 gov_raw.industry 列(英文id: power/cpi/pharma...) # 历史 bug: 曾从 daily_crawl_config.json 读 site_name→industry 映射, 但 config 无 industry 字段 → 全空 → 选任何行业都 0 # 行业过滤已下沉到 SQL（见各查询分支），不再对当前页内存过滤 —— 修复"选行业只有2条/0条" has_more = len(rows) > per if has_more: rows = rows[:per] db.close() elif db_type == 'eia': has_more, rows = False, [] if os.path.exists(EIA_DB_PATH): db = sqlite3.connect(EIA_DB_PATH); db.row_factory = sqlite3.Row c = db.cursor() fetch_size = per + 1 offset = (page-1)*per if q: inc, exc = parse_google_query(q) fts_parts = [] for t in inc: fts_parts.append(t) for t in exc: fts_parts.append('-' + t) if fts_parts: fts_q = ' AND '.join(fts_parts) offset = (page-1)*per c.execute("SELECT p.* FROM projects p JOIN projects_fts f ON p.id = f.rowid WHERE projects_fts MATCH ? ORDER BY rank LIMIT ? OFFSET ?", (sanitize_fts_query(fts_q), fetch_size, offset)) else: rows = [] else: offset = (page-1)*per c.execute("SELECT * FROM projects ORDER BY pub_date DESC, id DESC LIMIT ? OFFSET ?", (fetch_size, offset)) rows = [dict(r) for r in c.fetchall()]; db.close() has_more = len(rows) > per if has_more: rows = rows[:per] else: search_fn = znlh_search if db_type == 'znlh' else zc_search combined_q = q if q2: combined_q = combined_q + (' ' if combined_q else '') + q2 if q3: combined_q = combined_q + (' ' if combined_q else '') + q3 total, rows = search_fn(combined_q, page, per, daterange) # znlh/zc 保持原样（这些不频繁，暂时不动） has_more = False # has_more 用于"加载更多"按钮 def qp_extra(): parts = [] if q: parts.append('q='+urllib.parse.quote(q)) if q2: parts.append('q2='+urllib.parse.quote(q2)) if q3: parts.append('q3='+urllib.parse.quote(q3)) if daterange != 'all': parts.append('range='+daterange) if industry_filter: parts.append('industry='+urllib.parse.quote(industry_filter)) return '&'.join(parts) CSS = CSS_STYLE + """ .result-item .dl{font-size:13px;color:#5f6368;margin-top:2px} """ h = ''+title+' - BJIIR项目库' h += '' h += '
' # Nav (new style) h += '
' h += '' h += '
' for nav_id, href_path, nav_label in [('crawler', '/crawler/', '\u73af\u8bc4\u516c\u793a'), ('zc', '/zc/', '\u4e2d\u7b56\u5927\u6570\u636e'), ('cceup', '/ccpc', '\u4e2d\u9879\u7f51'), ('contact', '/contact', '\u8054\u7cfb\u4ebaDB'), ('originals', '/originals/', '\u539f\u59cb\u6570\u636e')]: act = ' class="act"' if nav_id == db_type else '' h += ''+nav_label+'' h += '
' h += '
'+self.username+' \u9000\u51fa
' h += '
' # Search section (new style) h += '
' main_form_action = '/'+db_type+'/' h += '
' h += '' h += '
' # Time range + extra fields sel_opts = {'all': '\u5168\u90e8\u65f6\u95f4', '1d': '\u8fc7\u53bb24\u5c0f\u65f6', '7d': '\u8fc7\u53bb\u4e00\u5468', '30d': '\u8fc7\u53bb\u4e00\u4e2a\u6708'} # Time range + extra fields sel_opts = {'all': '\u5168\u90e8\u65f6\u95f4', '1d': '\u8fc7\u53bb24\u5c0f\u65f6', '7d': '\u8fc7\u53bb\u4e00\u5468', '30d': '\u8fc7\u53bb\u4e00\u4e2a\u6708'} # 日期标签：一行全暴露（chips） h += '
' h += '时间' for dv, dl in sel_opts.items(): _cls = 'date-chip act' if daterange == dv else 'date-chip' h += f'' + dl + '' h += '
' # 行业标签：一行全暴露（chips） h += '
' h += '行业' # 全部行业 _cls = 'date-chip act' if not industry_filter else 'date-chip' h += f'全行业' for _ind in INDUSTRIES: _cls = 'date-chip act' if industry_filter == _ind['id'] else 'date-chip' h += f'' + _ind['cn'] + '' # other _cls = 'date-chip act' if industry_filter == 'other' else 'date-chip' h += f'其他' h += '
' # Total count h += '' + ('\u5171 ' + str(total) + ' \u6761\u7ed3\u679c \u00b7 ' if total > 0 else '') + '\u7b2c '+str(page)+' \u9875' # Results h += '
' for idx, r in enumerate(rows): pid = str(r.get('id') or r.get('project_id') or r.get('zid') or '') name = esc(r.get('project_name','') or r.get('owner_company','') or r.get('title','') or '') h += '
' h += '' + str(idx+1) + '' h += '
' h += '

' + name + '

' h += '
' h += '' + esc(r.get('publish_date', '') or '日期未知') + '' sn = esc(r.get('site_name', '') or '') if sn: h += '来源：' + sn + '' if db_type == 'crawler' and r.get('title'): _oid, _olevel = find_originals_id(r.get('title')) if _oid: _olbl, _otype = _orig_id_label(_oid) if _olbl: _ocls = 'orig-id' if _otype == 'proj' else 'orig-id comp' _otip = 'originals 项目级匹配 (Project ID)' if _otype == 'proj' else 'originals 公司级匹配 (Plant ID)' h += f'{_olbl}: {esc(str(_oid))}' if r.get('phase'): h += ''+esc(r['phase'])+'' if r.get('total_investment'): h += ''+esc(r['total_investment'])+'' elif r.get('budget'): h += ''+esc(r['budget'])+'' if r.get('province'): h += ''+esc(r['province'])+('/'+esc(r['city']) if r.get('city') else '')+'' h += '
' h += '
' h += '
' # Pagination if total > per: total_pages = max(1, (total + per - 1) // per) pag_params = {} if q: pag_params['q'] = q if q2: pag_params['q2'] = q2 if q3: pag_params['q3'] = q3 if daterange != 'all': pag_params['range'] = daterange if industry_filter: pag_params['industry'] = industry_filter h += self.paginate(page, total_pages, '/'+db_type+'/', pag_params) h += '
' self.send_response_html(200, h) def handle_db_detail(self, db_type, path): pid = path.split('/')[-1] _track_project_visit(db_type, pid, getattr(self, 'username', 'anonymous')) pv_all_visitors = [] pv_total = 0 if pid: try: dbt = sqlite3.connect(SESSION_DB) dbt.row_factory = sqlite3.Row c = dbt.execute('SELECT COUNT(*) FROM project_visits WHERE project_type=? AND project_id=?', (db_type, str(pid))) pv_total = c.fetchone()[0] c2 = dbt.execute('SELECT DISTINCT username, visited_at FROM project_visits WHERE project_type=? AND project_id=? ORDER BY visited_at DESC LIMIT 20', (db_type, str(pid))) pv_all_visitors = [dict(r) for r in c2] dbt.close() except: pass if db_type == 'crawler': p = crawler_detail(pid) names = {'crawler':'EIA'} elif db_type == 'eia': p = None if os.path.exists(EIA_DB_PATH): db = sqlite3.connect(EIA_DB_PATH); db.row_factory = sqlite3.Row c = db.execute("SELECT * FROM projects WHERE id=?", (pid,)) p = c.fetchone() db.close() names = {'eia':'EIA项目库'} if p: p = dict(p) else: detail_fn = znlh_detail if db_type == 'znlh' else zc_detail names = {'znlh':'中能联合','zc':'中策大数据'} p = detail_fn(pid) title = names.get(db_type, db_type) if not p: self.send_response_html(404, '

404

'); return h = ''+title+'' h += '' h += '← 返回列表' # Detail card header proj_name = esc(p.get('project_name', p.get('title', p.get('owner_company', '')))) h += '

' + proj_name + '

' if pv_total > 0: h += '' + str(pv_total) + ' 次访问' h += '
' # Render fields if db_type == 'znlh': for lbl,key in [('项目编号','project_id'),('当前阶段','phase'),('投资','budget'),('行业','industry'),('区域','region'),('省份','province'),('城市','city'),('工程类型','nature'),('占地面积','area'),('采购情况','purchase_status'),('竣工日期','completion_date'),('权重','weight'),('业主类型','owner_type'),('发布日期','publish_date'),('资金情况','funding_status')]: v = p.get(key,'') if v and v not in ('一一','——'): if key == 'industry': v = ind_cn(v) h += '
'+lbl+'：'+esc(v)+'
' else: for lbl,key in [('项目编号','project_id'),('版本类型','version_type'),('发布时间','publish_date'),('项目阶段','phase'),('建设周期','construction_period'),('总投资','total_investment'),('工程类型','project_type'),('甲方性质','owner_nature'),('行业','industry'),('规模','scale'),('省份','province'),('城市','city'),('详细地址','detail_address'),('建筑面积','building_area'),('占地面积','land_area'),('装修','decoration'),('电梯','elevator'),('空调','air_conditioning')]: v = p.get(key,'') if v and v not in ('一一','——'): if key == 'industry': v = ind_cn(v) h += '
'+lbl+'：'+esc(v)+'
' h += '
' # Crawler source info & original link if db_type == 'crawler': if p.get('publish_date'): h += '
发布日期：'+esc(p['publish_date'])+'
' if p.get('domain'): h += '
来源：'+esc(p['domain'])+'
' if p.get('url'): url_esc = esc(p['url']) h += '
原文链接：'+url_esc+'
' h += '' # Enterprise info (znlh) if db_type == 'znlh': for lbl,key in [('企业名称','owner_company'),('注册地址','registered_address'),('注册资金','registered_capital'),('企业负责人','company_contact'),('主营业务','business_scope')]: v = p.get(key,'') if v: h += '

'+lbl+'

'+esc(v)+'
' for lbl,key in [('建设背景','background'),('建设内容','construction_content'),('工艺路线','process_route'),('建设地点','construction_location'),('社会效益','social_benefits')]: v = p.get(key,'') if v: h += '

'+lbl+'

'+esc(v)+'
' for lbl,key in [('工艺设备','equipment_list'),('配套设施设备','supporting_equipment'),('客户所需设备','client_needed_equipment')]: v = p.get(key,'') if v: h += '

'+lbl+'

'+esc(v)+'
' if p.get('project_progress'): h += '

项目进展

'+esc(p['project_progress'])+'
' if p.get('detail'): h += '

特殊说明

'+esc(p['detail'])+'
' else: if p.get('construction_content'): h += '

建设内容

'+esc(p['construction_content'])+'
' if p.get('equipment_list'): h += '

设备清单

'+esc(p['equipment_list'])+'
' if p.get('schedule_overview'): h += '

工期概述

'+esc(p['schedule_overview'])+'
' # Crawler content - HTML to Markdown if db_type == 'crawler' and p.get('content'): h += ('

正文内容

' + (p['content'] or '') + '
' if '<' in (p['content'] or '') else '

正文内容

' + (p['content'] or '') + '
') # EIA content if db_type == 'eia': if p.get('pub_date'): h += '

发布日期

' + esc(p['pub_date']) + '
' if p.get('site_name'): h += '

来源站点

' + esc(p['site_name']) + '
' if p.get('url'): h += '

原文链接

' + esc(p['url']) + '
' if p.get('content'): h += ('

正文内容

' + (p['content'] or '') + '
' if '<' in (p['content'] or '') else '

正文内容

' + (p['content'] or '') + '
') # Contacts contacts = p.get('contacts',[]) if contacts: h += '

项目联系人

' for c in contacts: h += '
'+esc(c.get('company',''))+'
' if c.get('contact_name'): h += '姓名：'+esc(c['contact_name'])+'' if c.get('department'): h += '部门：'+esc(c['department'])+'' if c.get('position'): h += '职务：'+esc(c['position'])+'' if c.get('phone'): h += '手机：'+esc(c['phone'])+'' if c.get('address'): h += '地址：'+esc(c['address'])+'' if c.get('remarks') and c['remarks'] not in ('一一','——',''): h += '备注：'+esc(c['remarks'])+'' h += '
' h += '
' h += '' self.send_response_html(200, h) # ─── 搜索页面（gov FTS5 + categories）─── # ─── Google-style pagination ─── def paginate(self, page, total_pages, base_url, params): """Generate Google-style pagination: first/last pages + ellipsis + jump input. params is a dict of query params (without 'page'). """ if total_pages <= 1: url = "/?" + "&".join(f"{k}={urllib.parse.quote(str(v))}" for k, v in params.items() if v) return '' def url(p): all_params = {**params, 'page': str(p)} qs = '&'.join(f'{k}={urllib.parse.quote(str(v))}' for k, v in all_params.items() if v) return base_url + '?' + qs if qs else base_url html = '' return html def handle_search_page(self, params): q = self.get_param(params, 'q') q2 = self.get_param(params, 'q2', '').strip() q3 = self.get_param(params, 'q3', '').strip() sort = self.get_param(params, 'sort', 'date_desc') tag = self.get_param(params, 'tag', '') sub = self.get_param(params, 'sub', '') daterange = self.get_param(params, 'range', 'all') industry = self.get_param(params, 'industry', '') if industry not in INDUSTRY_MAP and industry != '': industry = '' page = int(self.get_param(params, 'page', '1')) page_size = int(self.get_param(params, 'page_size', '20')) if page_size not in (10, 20, 50, 100): page_size = 20 if sort not in ('relevance', 'date_desc', 'date_asc'): sort = 'date_desc' combined_q = q if q2: combined_q = combined_q + (' ' if combined_q else '') + q2 if q3: combined_q = combined_q + (' ' if combined_q else '') + q3 # Phase tabs: gov-only search if tag: if tag == '资源网站': sub_sources = {'中项网': cceup_search, '中能联合': znlh_search, '中策大数据': zc_search} if sub and sub in sub_sources: total, rows = sub_sources[sub](combined_q, page, page_size, daterange, industry) else: all_rows = [] for src_name, search_fn in [('中项网', cceup_search), ('中能联合', znlh_search), ('中策大数据', zc_search)]: t, r = search_fn(combined_q, 1, 10000, daterange, industry) for rr in r: rr['_source'] = src_name all_rows.extend(r) all_rows.sort(key=lambda x: x.get('publish_date','') or '', reverse=True) total = len(all_rows) offset = (page - 1) * page_size rows = all_rows[offset:offset+page_size] total_pages = max(1, (total + page_size - 1) // page_size) offset = (page - 1) * page_size html_page = self.render_main_page(q, rows, total, False, page, page_size, sort, tag, sub, offset, daterange, q2, q3, industry) self.send_response_html(200, html_page) return # Phase tag: gov search filtered by phase phase = tag if combined_q or daterange != 'all': results, total, has_more = search(combined_q, page, page_size, sort, phase, None, daterange, industry) else: results, total, has_more = [], 0, False offset = (page - 1) * page_size html_page = self.render_gov_page(q, results, total, has_more, page, page_size, sort, tag, offset, daterange, q2, q3, industry) self.send_response_html(200, html_page) return # Main page (no tag): search all DBs — 中项网 + 中策大数据 + EIA simultaneously all_rows = [] if combined_q or daterange != 'all' or (industry and industry != ''): # EIA gov # 行业筛选时需要全量取回（前200会被更新记录挤掉目标行业旧数据） # 2026-09-03: gov_entity 实体全称(公司/电话)命中时拉全量, 否则200截断会把正文命中丢光 _gq = (combined_q or '').strip() _gterms = _gq.split() _probe = None if len(_gterms) == 1 and ' ' not in _gq and not _gq.startswith('-'): _t = _gterms[0] if contains_chinese(_t) and len(_t) >= 3: _probe = ('company', _t) else: _dig = re.sub(r'\D', '', _t) if 7 <= len(_dig) <= 13: _probe = ('phone', _dig) _ecap = 0 if _probe: try: _dbp = get_db() _cp = _dbp.cursor() _pc = ['e.etype = ?', 'e.value = ?'] _pp = [_probe[0], _probe[1]] if daterange and daterange != 'all': _days = {'1d': 1, '7d': 7, '30d': 30}[daterange] _pc.append("(r.publish_date GLOB '20[0-9][0-9]-[0-9][0-9]-[0-9][0-9]*' AND SUBSTR(r.publish_date,1,10) >= date('now', '-%d days'))" % _days) if industry and industry != 'all': _pc.append('r.industry = ?') _pp.append(industry) _cp.execute('SELECT COUNT(*) FROM gov_entity e JOIN gov_raw r ON r.id = e.doc_id WHERE ' + ' AND '.join(_pc), _pp) _ecap = _cp.fetchone()[0] _dbp.close() except Exception: _ecap = 0 _gov_fetch = 10000 if (industry and industry != '') else (_ecap if _ecap else 200) gov_results, gov_total, _ = search(combined_q, 1, _gov_fetch, 'date_desc', None, None, daterange, industry) for rr in gov_results: rr['_source'] = 'EIA' # 实时打标（若 industry 列为空/other 则用标题推断并写回） if not rr.get('industry') or rr['industry'] == 'other': rr['_industry'] = classify_industry(rr.get('title','')) else: rr['_industry'] = rr['industry'] all_rows.extend(gov_results) # 中项网 _, cceup_rows = cceup_search(combined_q, 1, 200, daterange, industry) for rr in cceup_rows: rr['_source'] = '中项网' rr['_industry'] = classify_industry(rr.get('project_name','')) all_rows.extend(cceup_rows) # 中策大数据 _, zc_rows = zc_search(combined_q, 1, 200, daterange, industry) for rr in zc_rows: rr['_source'] = '中策大数据' rr['_industry'] = classify_industry(rr.get('project_name','')) all_rows.extend(zc_rows) # 若有industry筛选，对合并结果做二次过滤（兜底，主要针对EIA实时分类） if industry and industry != 'all': all_rows = [r for r in all_rows if r.get('_industry') == industry] all_rows.sort(key=lambda x: x.get('publish_date','') or '', reverse=True) has_more = len(all_rows) > page_size offset = (page - 1) * page_size rows = all_rows[offset:offset+page_size] if not rows and offset > 0: # 如果请求的页超出实际数据，回退到第一页 page = 1 offset = 0 rows = all_rows[:page_size] html_page = self.render_main_page(q, rows, len(all_rows), has_more, page, page_size, sort, '', '', offset, daterange, q2, q3, industry) self.send_response_html(200, html_page) def render_gov_page(self, query, results, total, has_more, page, page_size, sort, tag, offset, daterange, q2='', q3='', industry=''): def link_params(overrides): p = {'q': query, 'q2': q2, 'q3': q3, 'sort': sort, 'tag': tag or '', 'range': daterange, 'page': str(page), 'page_size': str(page_size)} if industry: p['industry'] = industry p.update({k: str(v) for k, v in overrides.items() if v}) return '&'.join(f'{k}={urllib.parse.quote(v)}' for k, v in p.items() if v) result_rows = '' for idx, r in enumerate(results): cat_badge = f'{esc(r.get("category",""))}' if r.get('category') else '' src_url = esc(r.get("page_url","") or r.get("source_url","") or '') tags_html = ' · '.join(PHASE_MAP.get(t,{}).get("label",t) for t in (r.get("tags") or [])) if r.get("tags") else '' result_rows += '
' result_rows += f'{offset + idx + 1}' result_rows += f'

{esc(r.get("title",""))}

' result_rows += '
' + cat_badge if tags_html: result_rows += f'{tags_html}' result_rows += f'{esc(r.get("publish_date","") or "日期未知")}' site_name = esc(r.get("site_name","") or "") if site_name: result_rows += f'来源：' + site_name + '' if r.get("title"): _oid, _olevel = find_originals_id(r.get("title")) if _oid: _olbl, _otype = _orig_id_label(_oid) if _olbl: _ocls = 'orig-id' if _otype == 'proj' else 'orig-id comp' _otip = 'originals 项目级匹配 (Project ID)' if _otype == 'proj' else 'originals 公司级匹配 (Plant ID)' result_rows += f'{_olbl}: {esc(str(_oid))}' result_rows += f'\U0001f4b0 {extract_investment(r.get("summary",""))}' result_rows += f'\U0001f517 \u539f\u6587' result_rows += '
' if not results and query: result_rows = '
\U0001f615 \u6ca1\u6709\u627e\u5230\u76f8\u5173\u7ed3\u679c
' elif not results and not query: result_rows = '
\u8f93\u5165\u5173\u952e\u8bcd\u5f00\u59cb\u641c\u7d22
' sel_all = ' selected' if daterange == 'all' else '' sel_1d = ' selected' if daterange == '1d' else '' sel_7d = ' selected' if daterange == '7d' else '' sel_30d = ' selected' if daterange == '30d' else '' sel_opts = {'all': '全部时间', '1d': '过去24小时', '7d': '过去一周', '30d': '过去一个月'} # 日期标签：一行全暴露（chips） def date_chip_url(dv): p = dict(params_dict) p['range'] = dv if dv != 'all' else '' p['page'] = '1' return '/?' + urllib.parse.urlencode({k: v for k, v in p.items() if v}) params_dict = {'q': query, 'sort': sort, 'tag': tag or '', 'range': daterange, 'q2': q2, 'q3': q3} if industry: params_dict['industry'] = industry date_bar = '
时间' for dv, dl in sel_opts.items(): _cls = 'date-chip act' if daterange == dv else 'date-chip' date_bar += f'{dl}' date_bar += '
' # Numbered pagination instead of load-more total_pages = max(1, (total + page_size - 1) // page_size) if total > 0 else 1 pagination = self.paginate(page, total_pages, '/', params_dict) # 行业筛选 chips def ind_chip_url(ind_id): p = dict(params_dict) if ind_id: p['industry'] = ind_id elif 'industry' in p: del p['industry'] p['page'] = '1' return '/?' + urllib.parse.urlencode({k: v for k, v in p.items() if v}) refine_html = '
行业' refine_html += f'全行业' for ind_id, label, kws in INDUSTRY_RULES: is_act = industry == ind_id refine_html += f'{label}' refine_html += f'其他' refine_html += '
' CSS = CSS_STYLE + """ .pagination{text-align:center;margin:20px 0;display:flex;justify-content:center;gap:4px;flex-wrap:wrap}.page-btn{display:inline-block;padding:8px 14px;background:#fff;border:1px solid #dadce0;border-radius:6px;font-size:14px;color:#1a73e8;text-decoration:none;transition:all .15s}.page-btn:hover{background:#e8f0fe;border-color:#1a73e8}.page-btn.active{background:#1a73e8;color:#fff;border-color:#1a73e8}.page-ellipsis{padding:8px 4px;color:#999}.page-jump{display:inline-flex;align-items:center;gap:4px;margin-left:8px}.page-jump input{width:50px;padding:4px 6px;border:1px solid #dadce0;border-radius:4px;text-align:center;font-size:13px} .meta .source{color:#0d652d;font-size:12px;margin-left:8px} """ return '''\n\n\n \n \n ''' + (esc(query) if query else '\u9879\u76ee\u4fe1\u606f\u68c0\u7d22') + '''\n \n\n\n
\n
环评公示 中策大数据 中项网 联系人DB 原始数据 原始数据
''' + self.username + ''' 退出
\n
\n \n \n
\n
\n
\n
''' + date_bar + '''
\n
\n \n
\n \n 共 ''' + str(total) + ''' \u6761\u7ed3\u679c\n
''' + refine_html + '''
''' + result_rows + '''
\n ''' + pagination + '''
\n\n''' def render_main_page(self, query, rows, total, has_more, page, page_size, sort, tag, sub, offset, daterange, q2='', q3='', industry=''): def link_params(overrides): p = {'q': query, 'q2': q2, 'q3': q3, 'sort': sort, 'tag': tag or '', 'sub': sub or '', 'range': daterange, 'page': str(page), 'page_size': str(page_size)} if industry: p['industry'] = industry p.update({k: str(v) for k, v in overrides.items() if v}) return '&'.join(f'{k}={urllib.parse.quote(v)}' for k, v in p.items() if v) result_rows = '' for idx, r in enumerate(rows): src = r.get('_source', '') pid = r.get('id') or r.get('project_id') or r.get('zid') or '' name = esc(r.get('project_name','') or r.get('owner_company','') or r.get('title','') or '') result_rows += '
' result_rows += f'{offset + idx + 1}' if src == 'EIA': detail_link = f'/crawler/project/{pid}' elif src == '中策大数据': detail_link = f'/zc/project/{pid}' else: detail_link = f'/ccpc/project/{pid}' result_rows += f'

{name}

' result_rows += '
' if src: result_rows += f'{esc(src)}' if r.get('budget'): result_rows += f'💰 {esc(str(r["budget"]))}' if r.get('province'): result_rows += f'📍 {esc(r["province"])}{"/"+esc(r["city"]) if r.get("city") else ""}' result_rows += f'📅 {esc(r.get("publish_date","") or "日期未知")}' result_rows += '
' if not result_rows: result_rows = '
😕 没有找到相关结果
' sel_opts = {'all': '全部时间', '1d': '过去24小时', '7d': '过去一周', '30d': '过去一个月'} params_dict = {'q': query, 'sort': sort, 'tag': tag or '', 'range': daterange, 'q2': q2, 'q3': q3} if industry: params_dict['industry'] = industry # 日期标签：一行全暴露（chips） def date_chip_url(dv): p = dict(params_dict) p['range'] = dv if dv != 'all' else '' p['page'] = '1' return '/?' + urllib.parse.urlencode({k: v for k, v in p.items() if v}) date_bar = '
时间' for dv, dl in sel_opts.items(): _cls = 'date-chip act' if daterange == dv else 'date-chip' date_bar += f'{dl}' date_bar += '
' # Numbered pagination instead of load-more total_pages = max(1, (total + page_size - 1) // page_size) if total > 0 else 1 pagination = self.paginate(page, total_pages, '/', params_dict) # 行业筛选 chips（一行全暴露） def ind_chip_url(ind_id): p = dict(params_dict) if ind_id: p['industry'] = ind_id elif 'industry' in p: del p['industry'] p['page'] = '1' return '/?' + urllib.parse.urlencode({k: v for k, v in p.items() if v}) industry_chips = '
行业' industry_chips += f'全行业' for ind_id, label, kws in INDUSTRY_RULES: is_act = industry == ind_id industry_chips += f'{label}' industry_chips += f'其他' industry_chips += '
' CSS = CSS_STYLE + """ .pagination{text-align:center;margin:20px 0;display:flex;justify-content:center;gap:4px;flex-wrap:wrap}.page-btn{display:inline-block;padding:8px 14px;background:#fff;border:1px solid #dadce0;border-radius:6px;font-size:14px;color:#1a73e8;text-decoration:none;transition:all .15s}.page-btn:hover{background:#e8f0fe;border-color:#1a73e8}.page-btn.active{background:#1a73e8;color:#fff;border-color:#1a73e8}.page-ellipsis{padding:8px 4px;color:#999}.page-jump{display:inline-flex;align-items:center;gap:4px;margin-left:8px}.page-jump input{width:50px;padding:4px 6px;border:1px solid #dadce0;border-radius:4px;text-align:center;font-size:13px} """ is_home = not query and not q2 and not q3 and not industry if is_home: return f''' BJIIR项目库
环保项目信息综合搜索
🔍 环评公示 📊 中策大数据 📄 中项网 👤 联系人DB 📋 原始数据
登录用户: {self.username}
''' else: return f''' {esc(query)} - BJIIR项目库
环评公示 中策大数据 中项网 联系人DB
{self.username} 退出
共 {total} 条结果
{date_bar}
{industry_chips}
{result_rows}
{pagination}
''' def render_cceup_page(self, title, category, q, rows, total, page, page_size, offset): tp = max(1, (total+page_size-1)//page_size) qp = '&q='+urllib.parse.quote(q) if q else '' h = ''+title+'' h += '
' h += '
' for nav_id, href_path, nav_label in [('zc', '/zc/', '中策大数据'), ('cceup', '/ccpc', '中项网'), ('contact', '/contact', '联系人DB'), ('originals', '/originals/', '原始数据'), ('eia', '/eia/', 'EIA项目库'), ('crawler', '/crawler/', 'EIA')]: h += ''+nav_label+'' h += ''+self.username+'' h += '登录用户 '+self.username+'退出
' h += '

'+title+'

' h += '
' h += '
共 '+str(total)+' 个项目
' for r in rows: pid = str(r.get('id') or r.get('project_id') or r.get('zid') or '') name = esc(r.get('project_name','') or r.get('owner_company','') or r.get('title','') or '') h += '
'+name+'
' if r.get('budget'): h += ''+esc(r['budget'])+'' if r.get('province'): h += ''+esc(r['province'])+('/'+esc(r['city']) if r.get('city') else '')+'' if r.get('publish_date'): h += ''+esc(r['publish_date'])+'' h += '
' if tp > 1: pag_params = {'category': '中项网'} if q: pag_params['q'] = q h += self.paginate(page, tp, '/', pag_params) h += '
' return h # ─── 原有CCEUP页面 ─── def handle_cceup_page(self, params): page = int(self.get_param(params, 'page', '1')) page_size = 50 q = self.get_param(params, 'q', '').strip() conn = sqlite3.connect(CONTACT_DB_PATH) conn.row_factory = sqlite3.Row c = conn.cursor() if q: c.execute("SELECT COUNT(*) FROM cceup_contacts WHERE company LIKE ? OR contact_name LIKE ? OR phone LIKE ? OR email LIKE ?", (f'%{q}%', f'%{q}%', f'%{q}%', f'%{q}%')) total = c.fetchone()[0] offset = (page - 1) * page_size c.execute("SELECT * FROM cceup_contacts WHERE company LIKE ? OR contact_name LIKE ? OR phone LIKE ? OR email LIKE ? ORDER BY id DESC LIMIT ? OFFSET ?", (f'%{q}%', f'%{q}%', f'%{q}%', f'%{q}%', page_size, offset)) else: c.execute("SELECT COUNT(*) FROM cceup_contacts") total = c.fetchone()[0] offset = (page - 1) * page_size c.execute("SELECT * FROM cceup_contacts ORDER BY id DESC LIMIT ? OFFSET ?", (page_size, offset)) rows = [dict(r) for r in c.fetchall()] conn.close() total_pages = max(1, (total + page_size - 1) // page_size) self.send_response_html(200, self.render_contact_list(q, rows, total, page, page_size, total_pages, offset)) def handle_contact_page(self, params): page = int(self.get_param(params, 'page', '1')) page_size = 50 q = self.get_param(params, 'q', '').strip() conn = sqlite3.connect(CONTACT_DB_PATH) conn.row_factory = sqlite3.Row c = conn.cursor() if q: c.execute("SELECT COUNT(*) FROM cceup_contacts WHERE company LIKE ? OR contact_name LIKE ? OR phone LIKE ? OR email LIKE ?", (f'%{q}%', f'%{q}%', f'%{q}%', f'%{q}%')) total = c.fetchone()[0] offset = (page - 1) * page_size c.execute("SELECT * FROM cceup_contacts WHERE company LIKE ? OR contact_name LIKE ? OR phone LIKE ? OR email LIKE ? ORDER BY id DESC LIMIT ? OFFSET ?", (f'%{q}%', f'%{q}%', f'%{q}%', f'%{q}%', page_size, offset)) else: c.execute("SELECT COUNT(*) FROM cceup_contacts") total = c.fetchone()[0] offset = (page - 1) * page_size c.execute("SELECT * FROM cceup_contacts ORDER BY id DESC LIMIT ? OFFSET ?", (page_size, offset)) rows = [dict(r) for r in c.fetchall()] conn.close() total_pages = max(1, (total + page_size - 1) // page_size) self.send_response_html(200, self.render_contact_list(q, rows, total, page, page_size, total_pages, offset)) def render_contact_list(self, q, rows, total, page, page_size, total_pages, offset): CSS = CSS_STYLE + ''' .contact-card{border:1px solid #e8e8e8;border-radius:8px;padding:16px;margin-bottom:12px;background:#fafafa;display:flex;align-items:flex-start;gap:16px} .contact-body{flex:1;min-width:0} .contact-dial{flex-shrink:0;display:flex;flex-direction:column;align-items:center;justify-content:center;padding:8px} .contact-dial a{display:flex;align-items:center;justify-content:center;width:48px;height:48px;border-radius:50%;background:#1a73e8;color:#fff;text-decoration:none;font-size:22px} .contact-dial a:hover{background:#1557b0} ''' h = '联系人DB - BJIIR项目库' h += '
' # Nav (new style) h += '
' h += '' h += '
' for nav_id, href_path, nav_label in [('crawler', '/crawler/', '\u73af\u8bc4\u516c\u793a'), ('zc', '/zc/', '\u4e2d\u7b56\u5927\u6570\u636e'), ('cceup', '/ccpc', '\u4e2d\u9879\u7f51'), ('contact', '/contact', '\u8054\u7cfb\u4ebaDB'), ('originals', '/originals/', '\u539f\u59cb\u6570\u636e')]: act = ' class="act"' if nav_id == 'contact' else '' h += ''+nav_label+'' h += '
' h += '
'+self.username+' \u9000\u51fa
' h += '
' # Search section h += '
' h += '
' h += '' h += '
' h += '
' h += '\u5171 '+str(total)+' \u6761\u8054\u7cfb\u4eba' h += '
' for idx, r in enumerate(rows): h += '
' h += '
' h += '
'+esc(r.get('company','') or '')+'
' h += '
' h += '\u8054\u7cfb\u4eba\uff1a'+esc(r.get('contact_name','') or '')+'' h += '\u624b\u673a\uff1a'+esc(r.get('phone','') or '')+'' if r.get('email'): h += '\u90ae\u7bb1\uff1a'+esc(r['email'])+'' if r.get('address'): h += '\u5730\u5740\uff1a'+esc(r['address'])+'' h += '
' phone = r.get('phone', '') if phone: h += '
\u260e
' h += '
' if total_pages > 1: pag_params = {} if q: pag_params['q'] = q h += self.paginate(page, total_pages, '/contact', pag_params) h += '
' return h def handle_cceup_projects(self, params): from html import escape page = int(self.get_param(params, 'page', '1')) q = self.get_param(params, 'q', '').strip() q2 = self.get_param(params, 'q2', '').strip() q3 = self.get_param(params, 'q3', '').strip() daterange = self.get_param(params, 'range', 'all') page_size = 50 combined_q = q if q2: combined_q = combined_q + (' ' if combined_q else '') + q2 if q3: combined_q = combined_q + (' ' if combined_q else '') + q3 total, rows = cceup_search(combined_q, page, page_size, daterange) total_pages = max(1, (total + page_size - 1) // page_size) CSS = CSS_STYLE + """ """ h = '中项网 - BJIIR项目库' h += '
' # Nav (new style) h += '
' h += '' h += '
' for nav_id, href_path, nav_label in [('crawler', '/crawler/', '\u73af\u8bc4\u516c\u793a'), ('zc', '/zc/', '\u4e2d\u7b56\u5927\u6570\u636e'), ('cceup', '/ccpc', '\u4e2d\u9879\u7f51'), ('contact', '/contact', '\u8054\u7cfb\u4ebaDB'), ('originals', '/originals/', '\u539f\u59cb\u6570\u636e')]: act = ' class="act"' if nav_id == 'cceup' else '' h += ''+nav_label+'' h += '
' h += '
'+self.username+' \u9000\u51fa
' h += '
' # Search section h += '
' h += '
' h += '' h += '
' # Time range sel_opts = {'all': '\u5168\u90e8\u65f6\u95f4', '1d': '\u8fc7\u53bb24\u5c0f\u65f6', '7d': '\u8fc7\u53bb\u4e00\u5468', '30d': '\u8fc7\u53bb\u4e00\u4e2a\u6708'} # 日期标签：一行全暴露（chips） def _date_chip_url(dv): _p = [] if q: _p.append('q=' + urllib.parse.quote(q)) if q2: _p.append('q2=' + urllib.parse.quote(q2)) if q3: _p.append('q3=' + urllib.parse.quote(q3)) if dv != 'all': _p.append('range=' + dv) return '/ccpc?' + '&'.join(_p) h += '
时间' for dv, dl in sel_opts.items(): _cls = 'date-chip act' if daterange == dv else 'date-chip' h += f'{dl}' h += '
' h += '
' h += '' + ('\u5171 ' + str(total) + ' \u6761\u7ed3\u679c \u00b7 ' if total > 0 else '') + '\u7b2c '+str(page)+' \u9875' h += '
' for idx, r in enumerate(rows): pid = str(r.get('id') or r.get('project_id') or r.get('zid') or '') name = escape(r.get('project_name','') or '') h += '
' h += '' + str(idx+1) + '' h += '
' h += '

'+name+'

' h += '
' if r.get('phase'): h += ''+escape(r['phase'])+'' if r.get('budget'): h += ''+escape(r['budget'])+'' if r.get('province'): h += ''+escape(r['province'])+('/'+escape(r['city']) if r.get('city') else '')+'' if r.get('publish_date'): h += ''+escape(r['publish_date'])+'' h += '
' h += '
' if total_pages > 1: pag_params = {} if q: pag_params['q'] = q if q2: pag_params['q2'] = q2 if q3: pag_params['q3'] = q3 if daterange != 'all': pag_params['range'] = daterange h += self.paginate(page, total_pages, '/ccpc', pag_params) h += '' self.send_response_html(200, h) def handle_cceup_project_detail(self, path): from html import escape pid = path.split('/')[-1] conn = sqlite3.connect(CONTACT_DB_PATH) conn.row_factory = sqlite3.Row r = conn.execute("SELECT * FROM cceup_projects WHERE id=?", (int(pid),)).fetchone() if not r: self.send_response_html(404, 'Not found'); return p = dict(r) contacts = [dict(r) for r in conn.execute("SELECT * FROM cceup_contacts WHERE project_id=?", (int(pid),))] conn.close() CSS = CSS_STYLE h = '\u4e2d\u9879\u7f51
''' h += '\u2190 \u8fd4\u56de\u5217\u8868' h += '
' + escape(p.get('project_name','') or '') + '
' # Basic info grid h += '

\u57fa\u672c\u4fe1\u606f

' fields = [ ('\u9879\u76ee\u7f16\u53f7','project_id'), ('\u9879\u76ee\u7c7b\u578b','project_type'), ('\u9879\u76ee\u9636\u6bb5','phase'), ('\u6295\u8d44\u91d1\u989d','budget'), ('\u884c\u4e1a','industry'), ('\u7ec6\u5206\u884c\u4e1a','field_type'), ('\u5de5\u7a0b\u7c7b\u578b','nature'), ('\u6295\u8d44\u6027\u8d28','invest_nature'), ('\u8d44\u91d1\u60c5\u51b5','funding'), ('\u9879\u76ee\u7b49\u7ea7','grade'), ('\u7701\u4efd','province'), ('\u57ce\u5e02','city'), ('地址','project_address'), ('\u53d1\u5e03\u65e5\u671f','publish_date'), ('\u7248\u672c','version'), ] for lbl, key in fields: v = p.get(key,'') if v and v not in ('\u4e00\u4e00','\u2014\u2014',''): h += '
' + lbl + '\uff1a' + escape(str(v)) + '
' h += '
' # Time and area info h += '

\u5de5\u671f\u4e0e\u89c4\u6a21

' for lbl, key in [('\u5f00\u5de5\u65f6\u95f4','start_time'),('\u5de5\u671f','end_time'),('\u5efa\u7b51\u9762\u79ef','building_area'),('\u5360\u5730\u9762\u79ef','land_area'),('\u8bbe\u5907\u6765\u6e90','equipment_source')]: v = p.get(key,'') if v and v not in ('\u4e00\u4e00','\u2014\u2014',''): h += '
' + lbl + '\uff1a' + escape(str(v)) + '
' h += '
' # Structure and fixtures info h += '

\u7ed3\u6784\u4e0e\u914d\u5957

' for lbl, key in [('\u94a2\u7ed3\u6784','steel_structure'),('\u5efa\u7b51\u5c42\u6570','building_floors'),('\u4f9b\u6696\u65b9\u5f0f','heating_method'),('\u5916\u5899\u6750\u6599','wall_material'),('\u88c5\u4fee','decoration'),('\u6709\u65e0\u7a7a\u8c03','has_ac'),('\u6709\u65e0\u7acb\u4f53\u505c\u8f66\u4f4d','has_parking'),('\u6709\u65e0\u7535\u68af','has_elevator')]: v = p.get(key,'') if v and v not in ('\u4e00\u4e00','\u2014\u2014',''): h += '
' + lbl + '\uff1a' + escape(str(v)) + '
' h += '
' # Text sections for lbl, key in [('项目概况','overview'),('工艺流程','process_flow'),('采购设备','procurement_equipment'),('采购需求','procurement_needs'),('项目进展','progress'),('设备清单','equipment_list'),('商机','opportunity'),('重点工作','key_works')]: v = p.get(key,'') if v and v.strip() and v not in ('一一','——'): h += '

' + lbl + '

' + escape(v) + '
' # Detail - render as key-value table detail_v = p.get('detail','') if detail_v and detail_v.strip() and detail_v not in ('一一','——'): lines = detail_v.split('\n') table_rows = [] text_parts = [] for line in lines: line = line.strip() if not line: continue if '：' in line: parts = line.split('：', 1) k = parts[0].strip() v = parts[1].strip() if k and v and k not in ('【所需材料设备】','【项目进展】','【项目详情】','版权所有'): table_rows.append((k, v)) continue text_parts.append(line) if table_rows: h += '

预测说明

' for k, v in table_rows: h += escape(k) + '：' + escape(v) + '\n' h += '
' if text_parts: h += '

项目详情

' + escape('\n'.join(text_parts)) + '
' # Owner company if p.get('owner_company'): h += '

业主

' + escape(p['owner_company']) + '
' # Contacts if contacts: h += '

联系人

' for c in contacts: phone = c.get('phone', '') or '' company = escape(c.get('company','') or '') contact_name = escape(c.get('contact_name','') or '') email = escape(c.get('email','') or '') address_val = escape(c.get('address','') or '') owner_esc = escape(p.get('owner_company','') or '') h += '
' if company: h += '
' + company + '
' h += '
' if contact_name: h += '姓名： ' + contact_name + '' if phone: h += '电话： ' + phone + '' if email: h += '邮箱： ' + email + '' h += '
' if address_val: h += '
地址：' + address_val + '
' h += '
' h += '
' h += '
' h += '
' self.send_response_html(200, h) def handle_detail(self, params): record_id = int(self.get_param(params, 'id', '0')) record = get_detail(record_id) if not record: self.send_response_html(404, 'Not found'); return from html import escape title = escape(record.get('title', 'Unknown')) site = escape(record.get('site') or '') cat = record.get('category') or '' date = record.get('publish_date') or '' status = record.get('status', '已采集') summary = record.get('summary', '') page_url = record.get('page_url') or '' source_url = record.get('source_url') or '' tags = record.get('tags') or '' if isinstance(tags, str): tags = [t.strip() for t in tags.split(',') if t.strip()] cat_html = f'{escape(cat)}' if cat else '' record_id_str = f'#{record_id}' summary_html = render_md(summary) if summary else '' content_raw = record.get('content', '') # Prefer clean HTML content; fall back to summary as markdown if content_raw and len(content_raw) > 50 and ('

' in content_raw or ' 50: # Treat as Markdown (pipe tables, etc.) and convert to HTML body_html = render_md(content_raw) elif summary and len(summary) > 50: body_html = summary_html else: body_html = '

正文内容未采集完整，请点击下方原文链接查看完整信息

' show_source = f'

📡 来源网站{site}

' if site else '' search_q = self.get_param(params, 'q', '') search_sort = self.get_param(params, 'sort', 'date_desc') search_phase = self.get_param(params, 'phase', '') search_category = self.get_param(params, 'category', '') search_page = self.get_param(params, 'page', '1') search_psize = self.get_param(params, 'page_size', '20') CSS = '''body{font-family:-apple-system,"Microsoft YaHei",sans-serif;background:#f5f5f5;margin:0;padding:0;color:#333} .container{max-width:800px;margin:0 auto;padding:20px} .nav-bar{margin-bottom:12px}.nav-bar a{color:#4361ee;text-decoration:none;font-size:14px} .detail-card{background:#fff;border-radius:12px;box-shadow:0 1px 4px rgba(0,0,0,.08);padding:24px} .detail-title{font-size:24px;margin-bottom:16px;color:#1a1a2e} .detail-meta{display:flex;flex-wrap:wrap;gap:12px;margin-bottom:20px;font-size:13px;color:#666;align-items:center} .detail-meta span{background:#f0f0f0;padding:4px 10px;border-radius:4px} .detail-status{color:#2e7d32;background:#e8f5e9!important} .detail-visits{color:#e65100} .detail-body{font-family:-apple-system,BlinkMacSystemFont,"Microsoft YaHei","PingFang SC",sans-serif;font-size:15px;line-height:1.6;color:#333} .detail-body p{margin:6px 0}.detail-body table{border-collapse:collapse;width:100%;margin:14px 0;font-size:14px}.detail-body td,.detail-body th{border:1px solid #ddd;padding:8px 12px;text-align:left;vertical-align:top}.detail-body tr:nth-child(even){background:#f9f9f9} .detail-source{border-top:1px solid #e5e7eb;padding-top:16px;margin-top:20px;font-size:14px} .detail-source-item{margin:8px 0} .detail-label{color:#888;margin-right:8px} .detail-original-link{color:#4361ee;word-break:break-all}''' h = f'''{title}
← 返回搜索结果

{title}

{cat_html}{site}📅 {date or "日期未知"}{status or "已采集"}👁 {record.get('visits',0) or 0} 次访问
{body_html}

🔗 原文链接{page_url or source_url}

{show_source}

数据ID{record_id_str}

''' self.send_response_html(200, h) def handle_visit(self, params): record_id = int(self.get_param(params, 'id', '0')) if record_id: conn = get_db() conn.execute("UPDATE gov_raw SET visits = visits + 1 WHERE id=?", (record_id,)) conn.commit(); conn.close() self.send_response_html(200, 'ok') def handle_dashboard(self): self.send_response_html(200, '

Dashboard

Coming soon

') def handle_sources_api(self): self.send_json({'sources': []}) def send_json(self, data): b = json.dumps(data, ensure_ascii=False).encode() self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(b))) self.end_headers() self.wfile.write(b) def log_message(self, format, *args): pass import json, sqlite3, os, subprocess, re CONFIG_PATH = "/root/gov_crawler/daily_crawl_config.json" SEARCH_DB = "/root/search.db" GOV_CRAWLER_DIR = "/root/gov_crawler" ADMIN_EXCLUDE_PATH = "/root/gov_crawler/admin_exclude_keywords.txt" INDUSTRIES = [ {"id": "power", "cn": "电力", "en": "Power"}, {"id": "terminals", "cn": "终端/接收站", "en": "Terminals"}, {"id": "port", "cn": "港口码头", "en": "Port & Harbor"}, {"id": "transmission", "cn": "管道输送", "en": "Transmission"}, {"id": "production", "cn": "生产/开采", "en": "Production"}, {"id": "altfuel", "cn": "替代燃料", "en": "Alternative Fuel"}, {"id": "hpi", "cn": "石油炼制", "en": "HPI (Hydrocarbon Processing Industry)"}, {"id": "cpi", "cn": "化工加工", "en": "CPI (Chemical Processing Industry)"}, {"id": "metals", "cn": "金属与矿物", "en": "Metals & Minerals"}, {"id": "pulp", "cn": "制浆/造纸/木材", "en": "Pulp, Paper & Wood"}, {"id": "food", "cn": "食品与饮料", "en": "Food & Beverage"}, {"id": "logistics", "cn": "物流", "en": "Logistics"}, {"id": "manufacturing", "cn": "工业制造", "en": "Industrial Manufacturing"}, {"id": "pharma", "cn": "制药与生物技术", "en": "Pharmaceutical & Biotech"}, ] def get_industries_json(): import json as _j return _j.dumps(INDUSTRIES, ensure_ascii=False) INDUSTRY_IDS = [ind["id"] for ind in INDUSTRIES] INDUSTRY_CN_MAP = {ind["id"]: ind["cn"] for ind in INDUSTRIES} def ind_cn(industry_id): """行业 id → 中文名（详情页显示用）；未知/other 保持原样""" if not industry_id: return '' return INDUSTRY_CN_MAP.get(industry_id, industry_id) _SCRIPT_URL_CACHE = {} _SCRIPT_SITE_CACHE = {} def extract_script_sitename(script_name): """Extract SITE_NAME constant from script source to match DB""" if not script_name: return "" for prefix in ["node ", "bash ", "python3 "]: if script_name.startswith(prefix): script_name = script_name[len(prefix):] break if script_name in _SCRIPT_SITE_CACHE: return _SCRIPT_SITE_CACHE.get(script_name, "") script_path = os.path.join(GOV_CRAWLER_DIR, script_name) if not os.path.isfile(script_path): _SCRIPT_SITE_CACHE[script_name] = "" return "" try: with open(script_path, "r", errors="replace") as f: head = f.read(5000) except: _SCRIPT_SITE_CACHE[script_name] = "" return "" m = re.search("^SITE_NAME\s*=\s*[\\\"']([^\\\"']+)[\\\"']", head, re.MULTILINE) if not m: m = re.search("^NAME\s*=\s*[\\\"']([^\\\"']+)[\\\"']", head, re.MULTILINE) if not m: m = re.search("^SITE\s*=\s*[\\\"']([^\\\"']+)[\\\"']", head, re.MULTILINE) sn = m.group(1) if m else "" _SCRIPT_SITE_CACHE[script_name] = sn return sn def extract_script_url(script_name): if not script_name: return "" for prefix in ["node ", "bash ", "python3 "]: if script_name.startswith(prefix): script_name = script_name[len(prefix):] break if script_name in _SCRIPT_URL_CACHE: return _SCRIPT_URL_CACHE.get(script_name, "") script_path = os.path.join(GOV_CRAWLER_DIR, script_name) if not os.path.isfile(script_path): _SCRIPT_URL_CACHE[script_name] = "" return "" try: with open(script_path, "r", errors="replace") as f: head = f.read(5000) except: _SCRIPT_URL_CACHE[script_name] = "" return "" # Priority 1: LIST_URL (most precise - list page URL) m = re.search("(?m)^(?:LIST_URL|list_url|listUrl)\\s*=\\s*[\"\'](https?://[^\"\']+)[\"\']", head) # Priority 2: API_URL if not m: m = re.search("(?m)^(?:API_URL|api_url|apiUrl)\\s*=\\s*[\"\'](https?://[^\"\']+)[\"\']", head) # Priority 3: PAGE_URL / SEARCH_URL / BASE_URL if not m: m = re.search("(?m)^(?:PAGE_URL|SEARCH_URL|search_url|BASE_URL)\\s*=\\s*[\"\'](https?://[^\"\']+)[\"\']", head) # Priority 4: BASE / base_url / DOMAIN if not m: m = re.search("(?m)^(?:BASE|base_url|DOMAIN)\\s*=\\s*[\"\'](https?://[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}[^\"\']*)[\"\']", head) # Priority 5: domain from docstring if not m: m = re.search(r"(https?://[a-zA-Z0-9.-]+\.[a-zA-Z]{2,})", head) url = m.group(1).rstrip("/") if m else "" _SCRIPT_URL_CACHE[script_name] = url return url def set_crawler_industry(name, industry_id): """Set industry for a crawler in config""" cfg = json.load(open(CONFIG_PATH)) if isinstance(cfg, list): cfg = {"crawlers": cfg} if industry_id and industry_id not in INDUSTRY_IDS: return {"success": False, "error": f"Unknown industry: {industry_id}"} for c in cfg["crawlers"]: if c["name"] == name: if industry_id: c["industry"] = industry_id else: c.pop("industry", None) json.dump(cfg, open(CONFIG_PATH, "w"), ensure_ascii=False, indent=2) return {"success": True, "name": name, "industry": industry_id or ""} return {"success": False, "error": f"Crawler \"{name}\" not found"} def get_crawler_stats(): """获取所有爬虫的状态数据""" cfg = json.load(open(CONFIG_PATH)) if isinstance(cfg, list): cfg = {"crawlers": cfg} crawlers = cfg['crawlers'] db = sqlite3.connect(SEARCH_DB) db.row_factory = sqlite3.Row # Query 1: GROUP BY site_name (fast, uses index) rows = db.execute(""" SELECT site_name, COUNT(*) as total, MAX(SUBSTR(publish_date,1,10)) as latest_date, MIN(SUBSTR(publish_date,1,10)) as oldest_date, SUM(CASE WHEN SUBSTR(publish_date,1,10) >= date('now', '-7 days') THEN 1 ELSE 0 END) as week_new, SUM(CASE WHEN SUBSTR(publish_date,1,10) >= date('now', '-1 days') THEN 1 ELSE 0 END) as day_new FROM gov_raw GROUP BY site_name """).fetchall() db_stats = {r['site_name']: dict(r) for r in rows} def normalize(s): return s.replace(' ', '').replace('—', '-').replace('–', '-').replace('（', '(').replace('）', ')').replace('　', '').replace(' ', '') db_stats_norm = {normalize(k): v for k, v in db_stats.items()} # Pre-warm cache: scan all script files once for SITE_NAME # (reuses _SCRIPT_SITE_CACHE populated by extract_script_sitename) # Build script_name set from DB (fast, index-only) _sn_set = set() try: _db_sn = __import__('sqlite3').connect(SEARCH_DB) for r in _db_sn.execute("SELECT DISTINCT script_name FROM gov_raw WHERE script_name IS NOT NULL AND script_name != ''").fetchall(): _sn_set.add(r[0]) _db_sn.close() except: _sn_set = set() db.close() result = [] for c in crawlers: name = c.get('name', c.get('site_name', c.get('script', ''))) script = c.get('script', c.get('command', '')) # 唯一标识: script_name字段 -> args里的--script-name -> 脚本文件名 _ck = c.get('script_name', '') if not _ck: _a = c.get('args', '') if isinstance(_a, list): _a = ' '.join(str(x) for x in _a) _m = re.search(r'--script-name=(\S+)', _a) if _a else None if not _m: _m = re.search(r'--script-name\s+(\S+)', _a) if _a else None if _m: _ck = _m.group(1) if not _ck: _ck = script ckey = _ck # Check if script file exists script_file = script for prefix in ['node ', 'bash ', 'python3 ']: if script.startswith(prefix): script_file = script[len(prefix):] break script_path = os.path.join(GOV_CRAWLER_DIR, script_file) if script_file else '' script_exists = os.path.isfile(script_path) if script_path else False script_name_key = '' script_name_full = '' if script: script_base = os.path.basename(script).replace('.py', '') script_name_full = script_base + '.py' if script_base.startswith('crawl_'): script_name_key = script_base[6:] else: script_name_key = script_base # Strategy 1 (now primary): match by DB script_name column -> site_name(s) -> db_stats # 有唯一标识(ckey, script_name字段/--script-name)的配置: 跳过脚本名聚合, # 直接走 site_name fallback 精确匹配(如江都13站各显示各站统计, 不合并) stats = {} if ckey and ckey != script: _keys_to_try = [] else: _keys_to_try = list(dict.fromkeys([k for k in [script_name_full, script_name_key] if k and k in _sn_set])) if _keys_to_try: try: _db_sn2 = __import__('sqlite3').connect(SEARCH_DB) _db_sn2.row_factory = __import__('sqlite3').Row _p = ','.join(['?' for _ in _keys_to_try]) _sites = set() for r in _db_sn2.execute(f"SELECT DISTINCT site_name FROM gov_raw WHERE script_name IN ({_p})", _keys_to_try).fetchall(): _sites.add(r['site_name']) _db_sn2.close() _total = 0 _latest = '-' _oldest = '-' _week_new = 0 _day_new = 0 for _site in _sites: _r = db_stats.get(_site) if _r: _total += _r['total'] if _r['latest_date'] and _r['latest_date'] > _latest: _latest = _r['latest_date'] if _r['oldest_date'] and (_oldest == '-' or _r['oldest_date'] < _oldest): _oldest = _r['oldest_date'] _week_new += _r['week_new'] _day_new += _r['day_new'] if _total > 0: stats = {'total': _total, 'latest_date': _latest, 'oldest_date': _oldest, 'week_new': _week_new, 'day_new': _day_new} except: pass if not stats: # Fallback: site_name matching (from pre-aggregated db_stats) stats = db_stats.get(name, {}) if not stats: # Try fuzzy match by normalized name stats = db_stats_norm.get(normalize(name), {}) if not stats: # Strategy: config name contains site_name name_norm = normalize(name) for sn, v in db_stats_norm.items(): if sn in name_norm: stats = v break if not stats: # Strategy: site_name contains config name for sn, v in db_stats_norm.items(): if name_norm in sn: stats = v break if not stats: # Strategy: shared prefix >= 3 chars for sn, v in db_stats_norm.items(): min_len = min(len(name_norm), len(sn)) if min_len >= 6: common = 0 for i in range(min_len): if name_norm[i] == sn[i]: common += 1 else: break if common >= 3: stats = v break if not stats: # Strategy: base name before separator base = name_norm.split('-')[0].split('—')[0].split('·')[0] if len(base) >= 2: for sn, v in db_stats_norm.items(): if base in sn or sn in base: stats = v break if not stats: # Strategy: strip common suffixes common_suffixes = ['通知公告', '公示公告', '环境保护', '环评信息', '环评公告', '环境信息', '政策文件', '新闻专区', '环评审批', '环境影响评价', '高新区动态', '公告公示', '生态环境保护'] base = name_norm for sfx in common_suffixes: if base.endswith(sfx): base = base[:-len(sfx)] break if len(base) >= 2 and base != name_norm: for sn, v in db_stats_norm.items(): if base in sn: stats = v break if not stats: # Strategy: match by script SITE_NAME constant site_name_from_script = extract_script_sitename(script_file) if site_name_from_script and site_name_from_script in db_stats: stats = db_stats[site_name_from_script] if not stats: # Strategy: match by script filename slug if script: script_base = os.path.basename(script).replace('.py', '') if script_base.startswith('crawl_'): slug = script_base[6:] for sn, v in db_stats.items(): if slug and (slug in sn or (len(slug) >= 3 and any(slug[:i] in sn for i in [4,6,8]))): stats = v break total = stats.get('total', 0) latest = stats.get('latest_date', '-') oldest = stats.get('oldest_date', '-') week_new = stats.get('week_new', 0) day_new = stats.get('day_new', 0) active_3d = active_7d = active_30d = active_180d = False if latest and latest != '-': try: from datetime import datetime, timedelta ld = datetime.strptime(latest[:10], '%Y-%m-%d') delta = datetime.now() - ld active_3d = delta <= timedelta(days=3) active_7d = delta <= timedelta(days=7) active_30d = delta <= timedelta(days=30) active_180d = delta <= timedelta(days=180) except: pass # ── 日跑状态: 从 run_logs 查今日状态 ── last_run_status = '⏳ 今日待运行' last_run_ts = 0 run_logs_history = [] try: _conn = __import__('sqlite3').connect('/root/search.db') _conn.row_factory = sqlite3.Row _today = datetime.now().strftime('%Y-%m-%d') # 按 script_name 精确匹配（优先唯一标识 ckey，兼容旧 script） _row = _conn.execute( "SELECT status, new_count, elapsed_seconds, error_detail, run_time " "FROM run_logs WHERE script_name=? AND run_date=?", (ckey, _today) ).fetchone() if not _row and ckey != script: _row = _conn.execute( "SELECT status, new_count, elapsed_seconds, error_detail, run_time " "FROM run_logs WHERE script_name=? AND run_date=?", (script, _today) ).fetchone() # script_name 没匹配到，试 config_name if not _row: _row = _conn.execute( "SELECT status, new_count, elapsed_seconds, error_detail, run_time " "FROM run_logs WHERE config_name=? AND run_date=?", (name, _today) ).fetchone() if _row: _st = _row['status'] _nc = _row['new_count'] if _st == '成功': last_run_status = f'✅今日成功 +{_nc}条' else: last_run_status = f'❌今日{_st}' # 最近5条历史（用于悬停） _hist = _conn.execute( "SELECT run_date, status, new_count FROM run_logs " "WHERE (script_name=? OR script_name=? OR config_name=?) AND run_date<=? " "ORDER BY run_date DESC LIMIT 5", (ckey, script, name, _today) ).fetchall() run_logs_history = [dict(h) for h in _hist] _conn.close() except: # 降级到 json 文件 try: with open('/root/gov_crawler/run_status.json') as _f: _status_map = __import__('json').load(_f) status_entry = _status_map.get(script) if not status_entry: status_entry = _status_map.get(name) if status_entry: last_run_ts = status_entry.get('ts', 0) status_str = status_entry.get('status', '') if status_str: last_run_status = status_str except: pass result.append({ 'name': name, 'script': script, 'group': c.get('group', '未分组'), 'enabled': c.get('enabled', True), 'note': c.get('note', ''), 'total': total, 'latest': latest, 'oldest': oldest, 'week_new': week_new, 'day_new': day_new, 'active': active_3d, 'active_3d': active_3d, 'active_7d': active_7d, 'active_30d': active_30d, 'active_180d': active_180d, 'script_exists': script_exists, 'last_run_status': last_run_status, 'last_run_ts': last_run_ts, 'run_logs_history': run_logs_history, }) return result, cfg def get_analysis_data(period='day', mode='incremental', metric='records', days=90): """ 获取分析数据 period: day/week/month mode: incremental / cumulative metric: records (数据量) / 脚本数 (脚本数) / both (双指标) days: 回溯天数 (0=全部) 脚本数按唯一脚本文件(.py)统计，脚本的新增日期取该脚本数据 在 search.db 中最早的 publish_date。 """ import json as _json import os as _os import datetime as _dt from collections import defaultdict as _dd db = sqlite3.connect(SEARCH_DB) db.row_factory = sqlite3.Row records_data = [] spiders_data = [] labels = [] cfg_path = "/root/gov_crawler/daily_crawl_config.json" # ====== 1. Build script-to-first-date mapping ====== # Get all unique site_names from DB with their first date db_site_dates = {} rows = db.execute(""" SELECT site_name, MIN(date(publish_date)) as d FROM gov_raw WHERE publish_date IS NOT NULL AND SUBSTR(publish_date,1,10) != '' AND date(publish_date) IS NOT NULL AND date(publish_date) != '1970-01-01' GROUP BY site_name """).fetchall() for r in rows: if r['site_name'] and r['d']: db_site_dates[r['site_name']] = r['d'] # Read daily_crawl_config.json, match each config entry to DB site_names scripts_map = {} # script_name -> earliest_date (YYYY-MM-DD or None) if _os.path.exists(cfg_path): try: cfg = _json.load(open(cfg_path)) if isinstance(cfg, list): cfg = {"crawlers": cfg} crawlers = cfg.get('crawlers', []) def normalize(s): return s.replace(' ', '').replace('—', '-').replace('–', '-') \ .replace('（', '(').replace('）', ')').replace('　', '').replace('\xa0', '') # Index normalized DB names db_norm = {normalize(k): k for k in db_site_dates} for c in crawlers: script = c.get('script', '') name = c.get('name', '') if not script or not name: continue earliest = None name_norm = normalize(name) # Strategy 0: site_names array if 'site_names' in c and isinstance(c['site_names'], list): for sn in c['site_names']: if sn in db_site_dates: d = db_site_dates[sn] if earliest is None or d < earliest: earliest = d # Strategy 1: exact match if earliest is None and name in db_site_dates: earliest = db_site_dates[name] # Strategy 2: normalized exact match if earliest is None and name_norm in db_norm: earliest = db_site_dates[db_norm[name_norm]] # Strategy 3: strip (#NNN) suffix if earliest is None: import re as _re clean = _re.sub(r'\(#\d+\)', '', name).strip() clean_norm = normalize(clean) if clean_norm in db_norm: earliest = db_site_dates[db_norm[clean_norm]] # Strategy 4: fuzzy - config name in DB site_name if earliest is None: for sn_norm, sn in db_norm.items(): if name_norm in sn_norm: d = db_site_dates[sn] if earliest is None or d < earliest: earliest = d # Strategy 5: DB site_name in config name if earliest is None: for sn_norm, sn in db_norm.items(): if sn_norm in name_norm: d = db_site_dates[sn] if earliest is None or d < earliest: earliest = d # Strategy 6: shared prefix >= 3 chars if earliest is None: for sn_norm, sn in db_norm.items(): min_len = min(len(name_norm), len(sn_norm)) if min_len >= 6: common = 0 for i in range(min_len): if name_norm[i] == sn_norm[i]: common += 1 else: break if common >= 3: d = db_site_dates[sn] if earliest is None or d < earliest: earliest = d # Strategy 7: base name before - if earliest is None: base = name_norm.split('-')[0].split('—')[0] if len(base) >= 2: for sn_norm, sn in db_norm.items(): if base in sn_norm: d = db_site_dates[sn] if earliest is None or d < earliest: earliest = d # Strategy 8: strip common suffixes if earliest is None: common_suffixes = ['通知公告', '公示公告', '环境保护', '环评信息', '环评公告', '环境信息', '政策文件', '环评审批', '环境影响评价', '公告公示', '行政许可公示'] base = name_norm for sfx in common_suffixes: if base.endswith(sfx): base = base[:-len(sfx)] break if len(base) >= 2 and base != name_norm: for sn_norm, sn in db_norm.items(): if base in sn_norm or sn_norm in base: d = db_site_dates[sn] if earliest is None or d < earliest: earliest = d scripts_map[script] = earliest if earliest is None: # Fallback: use file mtime try: mtime = _os.path.getmtime('/root/gov_crawler/' + script) scripts_map[script] = _dt.datetime.fromtimestamp(mtime).strftime('%Y-%m-%d') except: pass except Exception as e: print(f"get_analysis_data config error: {e}") # ====== 2. Compute records data from gov_raw ====== rec_where = "WHERE publish_date IS NOT NULL AND SUBSTR(publish_date,1,10) != '' AND date(publish_date) IS NOT NULL" if days > 0: rec_where += f" AND date(publish_date) >= date('now', '-{days} days')" if period == 'day': rec_date_expr = "date(publish_date)" rec_label_expr = "date(publish_date)" elif period == 'week': rec_date_expr = "date(publish_date, 'weekday 1', '-7 days')" rec_label_expr = "strftime('%Y-第%W周', publish_date)" else: rec_date_expr = "strftime('%Y-%m-01', publish_date)" rec_label_expr = "strftime('%Y年%m月', publish_date)" q_rec = f""" SELECT {rec_date_expr} as period_date, {rec_label_expr} as period_label, COUNT(*) as cnt FROM gov_raw {rec_where} GROUP BY period_date ORDER BY period_date ASC """ rec_rows = db.execute(q_rec).fetchall() rec_labels = [r['period_label'] for r in rec_rows] rec_inc = [r['cnt'] for r in rec_rows] # ====== 3. Compute spider data from scripts_map ====== # Filter scripts by first_date within range filtered_scripts = {} for s, d in scripts_map.items(): if d is None: # No data match - use file timestamp as proxy for "added date" try: mtime = _os.path.getmtime(f'/root/gov_crawler/{s}') d = _dt.datetime.fromtimestamp(mtime).strftime('%Y-%m-%d') except: continue if days > 0: cutoff = (_dt.datetime.now() - _dt.timedelta(days=days)).strftime('%Y-%m-%d') if d < cutoff: continue # Determine period label for this script's date if period == 'day': plabel = d elif period == 'week': dd = _dt.datetime.strptime(d, '%Y-%m-%d').date() monday = dd - _dt.timedelta(days=dd.weekday()) plabel = monday.strftime('%Y-第%W周') else: plabel = d[:7].replace('-', '年') + '月' filtered_scripts[s] = (d, plabel) # Count scripts per period spi_period_cnt = _dd(int) for s, (d, plabel) in sorted(filtered_scripts.items(), key=lambda x: x[1][0]): spi_period_cnt[plabel] += 1 spi_labels = sorted(spi_period_cnt.keys()) spi_inc = [spi_period_cnt[l] for l in spi_labels] # ====== 4. Merge records + 脚本数 ====== all_labels = sorted(set(rec_labels + spi_labels)) rec_inc_dict = dict(zip(rec_labels, rec_inc)) spi_inc_dict = dict(zip(spi_labels, spi_inc)) if mode == 'incremental': labels = all_labels records_data = [rec_inc_dict.get(l, 0) for l in all_labels] spiders_data = [spi_inc_dict.get(l, 0) for l in all_labels] else: labels = all_labels r_carry, s_carry = 0, 0 for l in all_labels: r_carry += rec_inc_dict.get(l, 0) s_carry += spi_inc_dict.get(l, 0) records_data.append(r_carry) spiders_data.append(s_carry) db.close() result = { 'labels': labels, 'record_counts': records_data, 'spider_counts': spiders_data if spiders_data else None, 'period': period, 'mode': mode, 'metric': metric, 'days': days, } return result def render_admin_page(): """渲染爬虫管理HTML页面""" crawlers, cfg = get_crawler_stats() total = len(crawlers) import sqlite3; total_records = sqlite3.connect(SEARCH_DB).execute("SELECT COUNT(*) FROM gov_raw").fetchone()[0]; sqlite3.connect(SEARCH_DB).close() active_3d = sum(1 for c in crawlers if c['active_3d']) active_7d = sum(1 for c in crawlers if c['active_7d']) active_30d = sum(1 for c in crawlers if c['active_30d']) active_180d = sum(1 for c in crawlers if c['active_180d']) groups = {} for c in crawlers: g = c['group'] if g not in groups: groups[g] = {'total': 0, 'records': 0, 'active': 0} groups[g]['total'] += 1 groups[g]['records'] += c['total'] if c['active']: groups[g]['active'] += 1 from datetime import datetime now_str = datetime.now().strftime('%Y-%m-%d %H:%M') week_total = sum(c['week_new'] for c in crawlers) day_total = sum(c['day_new'] for c in crawlers) parts = [] # 加载已保存的排除关键词 saved_exclude = "" try: with open(ADMIN_EXCLUDE_PATH) as f: saved_exclude = f.read().strip() except: pass parts.append(f''' 爬虫管理面板

🔍 爬虫管理面板

{total_records}
总数据条数
{total}
爬虫数
活跃站点（启用数）
近3日
{active_3d}
近7日
{active_7d}
近1月
{active_30d}
近半年
{active_180d}
+{day_total}
今日新增
+{week_total}
近7日新增
▶️
手动日跑
更新时间: {now_str}
关键词（逗号或空格分隔）：
''') for gname, gdata in sorted(groups.items()): parts.append(f' \n') parts.append('''
状态筛选:
''') for c in sorted(crawlers, key=lambda x: ((x.get('latest') or '')[::-1], -x.get('total',0))): active_marker = '🟢' if c['active'] else ('🟡' if c['week_new'] > 0 else '⚪') script_display = c['script'] if not c['script_exists']: script_display = f'⚠ {c["script"]}' elif not c.get('enabled', True): script_display = f'⏹️ {c["script"]}' # 从脚本源码提取真实请求URL（config url字段优先覆盖） _display_url = c.get('url', '') or c.get('source_url', '') if _display_url: url_cell = '%s' % (_display_url, _display_url, _display_url[:55]) else: url_cell = '-' run_status = c['last_run_status'] # 构建历史悬停信息 history_titles = [] for h in c.get('run_logs_history', []): _d = h['run_date'][:10] _s = '✅' if h['status'] == '成功' else '❌' _n = f"+{h['new_count']}条" if h['new_count'] else "" history_titles.append(f"{_d} {_s}{h['status']} {_n}") status_title = ' | '.join(history_titles[:5]) if history_titles else '' if run_status.startswith('⏳'): # 今日待运行 — 显示最近一次历史 if history_titles: status_title = f"最近: {history_titles[0]}" + (f"\n{history_titles[1]}" if len(history_titles) > 1 else "") # 颜色：新格式(✅今日/❌今日/⏳) + 旧格式(✅/⏰/🚫/❌)兼容 run_color = '#27ae60' if run_status.startswith('✅') else ( '#e67e22' if '⏰' in run_status or '超时' in run_status else ( '#8e44ad' if '🚫' in run_status or '不可达' in run_status else ( '#e74c3c' if run_status.startswith('❌') or '🐛' in run_status or '失败' in run_status or '异常' in run_status else '#999' ))) search_text = (c['name'] + ' ' + c['script'] + ' ' + (_display_url or '')).lower().replace('"', '') # Determine data-status for filtering if not c.get('enabled', True): _row_status = 'disabled' elif run_status.startswith('✅'): _row_status = 'success' elif '超时' in run_status: _row_status = 'timeout' elif '不可达' in run_status: _row_status = 'unreachable' elif run_status.startswith('❌') or '失败' in run_status or '异常' in run_status or '🐛' in run_status: _row_status = 'fail' else: _row_status = 'pending' _tr_extra = ' style="opacity:0.5"' if not c.get('enabled', True) else '' _enabled = c.get('enabled', True) _toggle_btn = ('') % ( str(c['name']).replace("'", "'"), 'background:#34c759;color:#fff;border-color:#34c759;" title="点击禁用"' if _enabled else 'background:#f5f5f5;color:#999;" title="点击启用"', '✅ 启用' if _enabled else '⏹️ 禁用', ) parts.append(f''' ''') parts.append('''
状态名称条数最新 7日新增活跃分组脚本来源URL日跑状态启用
{active_marker} {c['name']} {c['total']} {(c['latest'] or '')[:10] if (c['latest'] or '') != '-' else '-'} {c['week_new']} {'✅' if c['active'] else '❌'} {c['group']} {script_display} {url_cell} {run_status} {_toggle_btn}
数据量 脚本数
''') return ''.join(parts) def set_crawler_url(name, url): """设置爬虫的来源URL (保持原 flat-list 格式)""" cfg = json.load(open(CONFIG_PATH)) is_list = isinstance(cfg, list) crawlers = cfg if is_list else cfg.get('crawlers', []) for c in crawlers: cname = c.get('name', c.get('site_name', '')) if cname == name: if url: c['url'] = url else: c.pop('url', None) out = crawlers if is_list else cfg json.dump(out, open(CONFIG_PATH, 'w'), ensure_ascii=False, indent=2) return {'success': True, 'name': name, 'url': url} return {'success': False, 'error': f'Crawler "{name}" not found'} def toggle_crawler_enabled(name): """切换爬虫配置的启用/禁用状态 (保持原 flat-list 格式)""" cfg = json.load(open(CONFIG_PATH)) is_list = isinstance(cfg, list) crawlers = cfg if is_list else cfg.get('crawlers', []) for c in crawlers: cname = c.get('name', c.get('site_name', '')) if cname == name: cur = c.get('enabled', True) c['enabled'] = not cur out = crawlers if is_list else cfg json.dump(out, open(CONFIG_PATH, 'w'), ensure_ascii=False, indent=2) return {'success': True, 'name': name, 'enabled': c['enabled']} return {'success': False, 'error': f'Crawler "{name}" not found'} def run_single_script(name): """运行单个爬虫脚本""" import subprocess, json, os, time cfg = json.load(open(CONFIG_PATH)) if isinstance(cfg, list): cfg = {"crawlers": cfg} script = '' for c in cfg['crawlers']: if c.get('name', c.get('site_name', '')) == name: script = c.get('script', c.get('command', '')) break if not script: return {'success': False, 'error': 'Crawler "%s" not found' % name} # 构建命令 if script.endswith('.sh'): cmd = ['bash', os.path.join(GOV_CRAWLER_DIR, script)] elif script.endswith('.js'): cmd = ['node', os.path.join(GOV_CRAWLER_DIR, script)] else: cmd = ['python3', os.path.join(GOV_CRAWLER_DIR, script)] # 如果有 args，追加 for c in cfg['crawlers']: if c.get('name', '') == name: args = c.get('args', '') if args: if isinstance(args, list): cmd.extend([str(a) for a in args]) else: cmd.extend(str(args).split()) break try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) rc = result.returncode output = result.stdout[-500:] + result.stderr[-500:] # 更新日跑状态 now_ts = int(time.time()) run_record = { 'script': script, 'ts': now_ts, 'rc': rc, 'has_error': rc != 0, 'timeout': False, 'new_count': 0, 'name': name, } status_file = '/tmp/crawl_results.json' try: existing = json.load(open(status_file)) except: existing = [] existing.append(run_record) if len(existing) > 500: existing = existing[-500:] json.dump(existing, open(status_file, 'w')) # 也写入 run_status.json (兼容旧版) run_status_file = '/root/gov_crawler/run_status.json' run_emoji = '✅' if rc == 0 else '❌' try: rs = json.load(open(run_status_file)) except: rs = {} # Extract new_count from output if possible new_count = 0 for line in output.split('\n'): if '新增' in line or 'new' in line: import re m = re.search(r'(\d+)', line) if m: new_count = int(m.group(1)) rs[name] = {'status': '✅ +%d条' % new_count if rc == 0 else '❌ rc=%d' % rc, 'ts': now_ts} json.dump(rs, open(run_status_file, 'w'), ensure_ascii=False) # 写入 run_logs DB try: _db = __import__('sqlite3').connect('/root/search.db') _today = __import__('datetime').datetime.now().strftime('%Y-%m-%d') _now = __import__('datetime').datetime.now().strftime('%H:%M:%S') _status = '成功' if rc == 0 else '失败' _script_name = os.path.basename(script) if script else name _db.execute( "INSERT OR REPLACE INTO run_logs " "(run_date, run_time, script_name, config_name, status, new_count, elapsed_seconds, error_detail) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (_today, _now, _script_name, name, _status, new_count, 0, '' if rc == 0 else f'rc={rc}')) _db.commit() _db.close() except: pass return { 'success': rc == 0, 'name': name, 'script': script, 'rc': rc, 'output': output[:200], 'message': '已完成 (exit 0)' if rc == 0 else 'rc=%d' % rc } except subprocess.TimeoutExpired: return {'success': False, 'error': '执行超时(300s)', 'name': name} except Exception as e: return {'success': False, 'error': str(e), 'name': name} # ─── 手动日跑 ─── RUN_PID_FILE = "/tmp/daily_crawl_run.pid" RUN_LOG_TAIL = 20 def run_daily_crawl(): import os, subprocess status = get_daily_run_status() if status.get('running'): return {'success': False, 'message': '日跑已在运行中', 'running': True} try: pid = os.fork() if pid == 0: os.setsid() subprocess.run(['bash', '/root/daily_incremental.sh'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: os.remove(RUN_PID_FILE) except: pass os._exit(0) else: with open(RUN_PID_FILE, 'w') as f: f.write(str(pid)) return {'success': True, 'message': '日跑已启动', 'running': True, 'pid': pid} except Exception as e: return {'success': False, 'message': f'启动失败: {e}', 'running': False} def get_daily_run_status(): import os, subprocess if os.path.exists(RUN_PID_FILE): try: with open(RUN_PID_FILE) as f: pid = int(f.read().strip()) os.kill(pid, 0) log_tail = '' try: log_tail = subprocess.check_output( ['tail', f'-{RUN_LOG_TAIL}', '/var/log/crawler_daily.log'], timeout=3, stderr=subprocess.DEVNULL).decode('utf-8', errors='replace') except: pass return {'running': True, 'pid': pid, 'log_tail': log_tail} except (OSError, ValueError): try: os.remove(RUN_PID_FILE) except: pass return {'running': False} if __name__ == '__main__': http.server.ThreadingHTTPServer(('0.0.0.0', PORT), SearchHandler).serve_forever()