#!/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"
# 2026-09-10: 统一联系人库(contact_lib.py 从 ccpc.db+ZC.db 聚合) — /contact 页专用;
# CONTACT_DB_PATH(ccpc.db) 仍服务 /ccpc 项目浏览与项目详情联系人
CONTACT_ROUTE_DB = "/root/contact.db"
# 角色展示归一/过滤(2026-09-10 用户裁定): '业主/业主单位/owner' 是硬模板默认值不显示;
# 英文别名归一到中文; 过滤后为空则不渲染"角色"字段
CONTACT_TEMPLATE_ROLES = {'业主', '业主单位', '业主方', 'owner', 'ref'}
CONTACT_ROLE_ALIAS = {'owner': '业主', 'design': '设计院', 'contractor': '施工单位'}
def contact_role_display(raw):
parts = [p.strip() for p in (raw or '').split(' / ') if p.strip()]
out = []
for p in parts:
v = CONTACT_ROLE_ALIAS.get(p.lower(), p)
if v in CONTACT_TEMPLATE_ROLES or v.lower() in CONTACT_TEMPLATE_ROLES:
continue
if v not in out:
out.append(v)
return ' / '.join(out)
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'(?），此处不再重复定义 */
.topbar{display:flex;align-items:center;padding:12px 0 0}
.search-section{margin-bottom:12px}
.search-box{display:flex;gap:0}
.search-box input{flex:1;padding:12px 20px;border:1px solid #dfe1e5;border-radius:24px 0 0 24px;font-size:16px;outline:none;box-shadow:0 1px 6px rgba(32,33,36,.08)}
.search-box input:focus,.search-box input:hover{border-color:#1a73e8;box-shadow:0 1px 8px rgba(32,33,36,.16)}
.search-box button{padding:12px 24px;background:#1a73e8;color:#fff;border:none;border-radius:0 24px 24px 0;font-size:15px;cursor:pointer}
.search-box button:hover{background:#1557b0}
.toolbar{margin-bottom:10px}
.controls{display:flex;justify-content:space-between;align-items:center}
.total-count{font-size:14px;color:#70757a}
.db-tabs{display:flex;gap:6px;margin-bottom:12px;flex-wrap:wrap}
.db-tabs a{font-size:13px;color:#5f6368;padding:5px 14px;border:1px solid #dadce0;border-radius:16px;transition:all .15s}
.db-tabs a:hover{background:#e8f0fe;text-decoration:none;border-color:#1a73e8;color:#1a73e8}
.db-tabs a.act{background:#1a73e8;color:#fff;border-color:#1a73e8}
.results{max-width:100%;margin:0 auto}.result-item{display:flex;gap:10px;padding:12px 0;border-bottom:1px solid #f0f0f0;transition:background .1s}
.result-item:hover{background:#f8f9fa;margin:0 -10px;padding:12px 10px;border-radius:4px}
.result-item .num{font-size:13px;color:#70757a;min-width:26px;text-align:right;flex-shrink:0;padding-top:2px}
.result-item h3{font-size:16px;font-weight:400;line-height:1.3;margin-bottom:2px}
.result-item h3 a{color:#1a0dab}
.meta{font-size:12px;color:#70757a;display:flex;gap:8px;flex-wrap:wrap;align-items:center}
.bdg{display:inline-block;padding:1px 6px;border-radius:3px;font-size:11px;font-weight:500}
.bdg-crawler{background:#e8f5e9;color:#2e7d32}
.bdg-zc{background:#fef3e2;color:#e65100}
.bdg-other{background:#f3e8fd;color:#7b1fa2}
.orig-id{display:inline-block;padding:1px 6px;border-radius:3px;font-size:11px;font-weight:600;background:#fff3e0;color:#e65100;cursor:help}
.orig-id.comp{background:#f3e8fd;color:#7b1fa2;font-weight:500}
.summary{font-size:14px;color:#4d5156;line-height:1.58;margin-top:2px}
.no-results{text-align:center;padding:60px 0;color:#70757a;font-size:14px}
.time-select{padding:5px 12px;font-size:13px;color:#3c4043;border:1px solid #dadce0;border-radius:16px;cursor:pointer;outline:none;background:#fff}
.time-select:hover{border-color:#bdc1c6}
.page-btn{display:inline-flex;align-items:center;justify-content:center;min-width:36px;height:36px;padding:0 12px;border-radius:4px;font-size:13px;color:#1a73e8;text-decoration:none}
.page-btn:hover{background:#f0f0f0;text-decoration:none}
.page-btn.act{background:#1a73e8;color:#fff}
.page-elp{color:#70757a;padding:0 4px;font-size:13px}
.page-jump{display:inline-flex;align-items:center;gap:4px;margin-left:6px;font-size:13px;color:#5f6368}
.page-jump input{width:48px;padding:6px 8px;border:1px solid #dadce0;border-radius:4px;font-size:13px;text-align:center;outline:none}
.page-jump input:focus{border-color:#1a73e8}
.detail-wrap{max-width:900px;margin:0 auto;padding:20px}
.detail-card{background:#fff;border-radius:12px;padding:24px;margin-bottom:16px;box-shadow:0 1px 3px rgba(0,0,0,.08)}
.detail-card h2{font-size:20px;margin-bottom:12px;color:#1a1a2e}
.detail-content{font-size:14px;line-height:1.8;color:#444;overflow-wrap:break-word;overflow-x:auto}
.detail-content p{margin:10px 0}
.detail-content table{border-collapse:collapse;width:100%;margin:12px 0;font-size:14px}
.detail-content td,.detail-content th{border:1px solid #ddd;padding:8px 12px}
.detail-content a{color:#1a73e8}
.detail-content img{max-width:100%;height:auto;border-radius:8px}
/* ── 2026-09-11 移动端断点（首页 + 搜索结果页是本站唯一没有断点的页面）──
修复实测缺陷：375px 下 .nav-right（247px 宽）造成横向溢出 6px、
.nav-links a 被挤到高 80~98px（导航文字逐字换行）。 */
@media(max-width:760px){
.container{padding:10px 12px 40px}
.nav{flex-wrap:wrap;gap:8px;padding:8px 0;margin-bottom:12px}
.nav-logo{font-size:17px}
.nav-links{gap:2px;width:100%}
.nav-links a{font-size:12.5px;padding:5px 9px;white-space:nowrap;flex:0 0 auto}
.topbar{padding:8px 0 0}
h1{font-size:20px}
.home-logo{font-size:34px}
.home-sub{margin-bottom:20px}
.home-wrap{min-height:60vh}
.search-box input{font-size:16px;padding:12px 15px}
.search-box button{padding:12px 16px;font-size:14px}
.controls{flex-direction:column;align-items:flex-start;gap:8px}
.total-count{font-size:13px;line-height:1.6}
.date-bar{flex-wrap:wrap;gap:4px}
.date-chip{font-size:12px;padding:4px 9px}
.result-item{padding:10px 0}
.result-item .num{min-width:20px}
.result-item h3{font-size:15px}
.meta{font-size:11.5px;gap:6px}
.pagination{padding:12px 10px;gap:4px;margin:20px 0 44px}
.page-btn{min-width:34px;height:34px;padding:0 9px;font-size:12.5px}
.page-jump{margin-left:0}
.pager-bar{padding:10px 12px}
}
"""
def token_grams(t):
"""提取 token 的中文连续段的 2-gram 列表（标点/数字/字母作为分隔符跳过）。
返回空列表 = 该 token 无有效中文 gram（如纯英文/单字中文），无法用 bigram 索引。"""
segs = re.findall(r'[\u4e00-\u9fff]+', t)
grams = []
for seg in segs:
if len(seg) == 1:
continue
grams.extend(seg[i:i + 2] for i in range(len(seg) - 1))
return grams
def parse_google_query(q):
"""解析 Google 语法查询：-keyword 排除（仅当 - 前有空格/独立成词），其余为包含。
标点（- : () 等）前面无空格时视为关键词的一部分，不触发高级搜索语法。"""
if not q:
return [], []
q = normalize_query(q)
parts = q.strip().split()
inc, exc = [], []
for p in parts:
if p.startswith("-") and len(p) > 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', 'bjiir'}
NAV_CLUSTER_CSS = (
''
)
# 统一分页条样式（2026-09-10）—— 自包含，paginate() 与 admin 脚本表共用。
# 设计要点（用户反馈「底部的页码与分页按钮太靠底部了，不容易看清楚」）：
# ① 白色卡片 + 边框 + 投影 → 与页面背景分离，醒目
# ② margin-bottom 60px → 不再贴着窗口底边
# ③ 38px 高按钮 + hover/active 态 → 好点、易看清
PAGER_CSS = (
''
)
def nav_right_html(username, fixed=False):
"""右上角导航簇（2026-09-10 用户需求；样式自包含，不依赖页面 CSS）。
- 「登录用户」从主页面居中位置移到右上角，并新增「退出」按钮（主页面此前无退出）
- 「TM Script Center」→ /tm（反代 127.0.0.1:8010）；「Admin」→ /crawler/admin
- Admin 可见性：**除 bjiir 外所有登录用户可见**（bjiir 看不到；用户 2026-09-10 确认）
- 工具市场可见性：**同样对 bjiir 隐藏**（2026-09-11 用户需求；与 Admin 同一口径 = 入口级隐藏，
路由不做角色校验，与 Admin 现状一致）
- 顺位：Admin · 工具市场 · TM Script Center · 登录用户 · 退出（Admin 在 TM 入口左侧）
- fixed=True → 常驻浏览器窗口右上角（主页面用）；否则在流内右对齐，
且仅在视口 ≥1760px 时脱离文档流贴窗口右上角（避免与页内 tab 行重叠）
"""
u = username or ''
parts = []
if u and u != 'bjiir':
parts.append('Admin')
parts.append('工具市场')
parts.append('TM Script Center')
parts.append('' + u + '')
parts.append('退出')
cls = 'nav-right fixed' if fixed else 'nav-right'
return NAV_CLUSTER_CSS + '
' + ''.join(parts) + '
'
SESSION_DB = '/root/session.db' # 2026-09-10: 会话/访问统计迁出 contact.db(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 ───
# 正文是不是"真 HTML"。
# ⚠️ 绝对不能用 `'<' in content` 判断 —— 正文里出现书名号嵌套（《…<…>…》，
# 如《安徽省大气办关于印发<安徽省2020年大气污染防治重点工作任务>的通知》）
# 或数学比较符（COD<50）都会**误判成 HTML**，然后落进 .ct-html（它没有
# white-space:pre-line）→ 正文里所有 \n\n 被当空白吞掉 → 整篇塌成一坨。
# 2026-09-11 实例：广德市人民政府-环评审批 id=2097613798365819047，5190 字正文
# 里只有 1 个 '<'（就是上面那个书名号嵌套），整篇无分段。
_BODY_TAG_RE = re.compile(r']*)?/?>')
# ─── 正文噪声块剥离（2026-09-11）─────────────────────────────────────
# 政府站常把整段内联 CSS/JS 塞在正文开头。实例（绍兴上虞区政府-环评公示，
# id=2097613798365821229）的 content 开头就是：
#