#!/usr/bin/env python3
"""
线索评价分析报告 — 跨数据源统一分析
输出：HTML 报告（高价值线索 + 三星线索，每条带上传按钮）
"""
import json
import os
import re
from datetime import datetime, timedelta
# ============ 评价规则 ============
CLIENT_GRADING = {
"电建十一局": "S", "河南工程局": "S", "电建漯河分公司": "S",
"电建一公司": "S", "电建二公司": "S", "新疆九峰山": "S",
"电建四公司": "S", "电建三公司": "S",
"新矿集团": "A", "电建物资公司": "A", "焦作分公司": "A",
"广西分公司": "B", "贵州工程局": "B",
}
CLIENT_SCORE = {"S": 20, "A": 10, "B": 0}
CORE_CATEGORIES = ["防火门", "配电箱", "电线电缆", "涂料", "阀门", "瓷砖", "钢材", "防水材料", "水泥", "混凝土", "钢筋"]
PREFERRED_PROJECTS = ["抽水蓄能", "光伏", "学校", "医院", "河道治理", "水利", "新能源", "医养", "养老", "公路", "矿山"]
AVOID_WORDS = ["商品住宅", "房地产", "住宅楼", "商品房"]
def today_dir():
return datetime.now().strftime("~/desktop/招标公告/%Y-%m-%d")
def score_by_client(org_name):
for keyword, level in CLIENT_GRADING.items():
if keyword in org_name:
return level
return None
def score_by_category(title):
return [cat for cat in CORE_CATEGORIES if cat in title]
def score_by_project_type(title):
return [proj for proj in PREFERRED_PROJECTS if proj in title]
def score_risk_check(title):
return [word for word in AVOID_WORDS if word in title]
def estimate_amount_value(title, org):
text = title + org
value = 0
scale_hints = ["大型", "枢纽", "水库", "主干道", "高架", "跨江", "跨河", "隧道", "地铁", "水电站"]
for hint in scale_hints:
if hint in text:
value += 10
break
if "采购" in title:
value += 5
for tag in ["万吨", "万方", "万块", "km", "公里", "批量"]:
if tag in title:
value += 10
break
if any(k in title for k in ["几内亚", "海外", "国外", "境外", "国际"]):
value += 5
return value
def evaluate_lead(title, org):
reasons = []
total = 50
client_level = score_by_client(org)
if client_level:
add = CLIENT_SCORE[client_level]
if add > 0:
total += add
reasons.append(f"甲方{ {'S':'S级','A':'A级','B':'B级','C':'C级'}.get(client_level,client_level)}+{add}")
else:
reasons.append("甲方待核实")
cat_matches = score_by_category(title)
if cat_matches:
total += 15
reasons.append(f"品类匹配+15({','.join(cat_matches)})")
else:
reasons.append("品类未命中")
proj_matches = score_by_project_type(title)
if proj_matches:
total += 10
reasons.append(f"偏好项目+10({','.join(proj_matches)})")
amount_score = estimate_amount_value(title, org)
if amount_score:
total += amount_score
reasons.append(f"金额评估+{amount_score}")
risk_hits = score_risk_check(title)
if risk_hits:
total -= 20
reasons.append(f"风险扣分-20({','.join(risk_hits)})")
total = max(total, 0)
if total >= 90:
star = 5
elif total >= 80:
star = 4
elif total >= 70:
star = 3
elif total >= 55:
star = 2
else:
star = 1
return {"score": total, "star": star, "reasons": reasons}
def load_notices():
notices = []
base_dir = os.path.expanduser(today_dir())
if not os.path.exists(base_dir):
return notices
for fname in os.listdir(base_dir):
if fname.startswith("公告清单_水电十一局_") and fname.endswith(".md"):
with open(os.path.join(base_dir, fname)) as f:
content = f.read()
for block in re.split(r"(?=^## \d+\.)", content, flags=re.MULTILINE):
if not block.strip() or block.startswith("# "):
continue
m = re.search(r"## \d+\.\s+(.+)", block)
if not m:
continue
title = m.group(1).strip()
pub = re.search(r"\*\*发布日期\*\*：`(.+)`", block)
dl = re.search(r"\*\*投标截止\*\*：`(.+)`", block)
nid = re.search(r"\*\*公告ID\*\*：`(\d+)`", block)
url_m = re.search(r"\*\*详情链接\*\*：(.+)", block)
notices.append({
"title": title, "org": "电建十一局",
"pub_time": pub.group(1).strip() if pub else "",
"source": "PowerChina",
"deadline": dl.group(1).strip() if dl else "",
"url": url_m.group(1).strip() if url_m else "",
"cost_type": "招采", "_notice_id": nid.group(1) if nid else "",
})
break
for fname in os.listdir(base_dir):
if fname.startswith("公告清单_新矿集团_") and fname.endswith(".md"):
with open(os.path.join(base_dir, fname)) as f:
content = f.read()
for block in content.split("---\n"):
if not block.strip() or block.startswith("# "):
continue
m = re.search(r"## \d+\.\s+(.+)", block)
if not m:
continue
title = m.group(1).strip()
org = (re.search(r"\*\*发布组织\*\*:\s*(.+)", block) or [None, ""]).group(1).strip()
pt = (re.search(r"\*\*发布时间\*\*:\s*(.+)", block) or [None, ""]).group(1).strip()
nt = (re.search(r"\*\*类型\*\*:\s*(.+)", block) or [None, ""]).group(1).strip()
url = (re.search(r"\[查看\]\((.+)\)", block) or [None, ""]).group(1).strip()
notices.append({
"title": title, "org": org, "pub_time": pt,
"source": "新矿集团", "deadline": "", "url": url,
"cost_type": nt,
})
break
return notices
def generate_html_report():
print("正在加载数据...")
all_notices = load_notices()
print(f" 共 {len(all_notices)} 条")
print("\n正在评价...")
evaluated = [{"_id": i, **n, **evaluate_lead(n["title"], n["org"])} for i, n in enumerate(all_notices)]
today = datetime.now()
cutoff = today - timedelta(days=7)
evaluated = [e for e in evaluated if e["pub_time"] and (lambda t: (datetime.strptime(t[:10], "%Y-%m-%d") >= cutoff) if t else False)(e["pub_time"])]
evaluated.sort(key=lambda x: x["score"], reverse=True)
high_value = [e for e in evaluated if e["star"] >= 4]
star3 = [e for e in evaluated if e["star"] == 3]
print(f" 高价值线索: {len(high_value)} 条")
print(f" 三星线索: {len(star3)} 条")
today_str = today.strftime("%Y-%m-%d")
def card(item):
star_icons = "★" * item["star"] + "☆" * (5 - item["star"])
reasons = " · ".join(item["reasons"])
url = item.get("url", "")
title_html = f'{item["title"]}' if url else item["title"]
is_high = item["star"] >= 4
today_flag = '今日' if item.get("pub_time", "")[:10] == today_str else ""
# 构建上传数据 JSON
cat = next((c for c in CORE_CATEGORIES if c in item["title"]), "通用")
up_data = json.dumps({
"title": item["title"][:50], "deadline": (today + timedelta(days=14)).strftime("%Y-%m-%d"),
"region": item.get("org", ""), "category": cat, "unit": "批", "quantity": "1",
}, ensure_ascii=False)
upload_btn = f'' if not is_high else ''
return f'''
{item["org"]} · {item.get("cost_type","")} · {item.get("deadline") or item.get("pub_time","")[:10]}
{reasons}
'''
html = f'''
线索综合分析报告 - {today_str}
线索综合分析报告
{today_str} | PowerChina ({len([n for n in evaluated if n["source"]=="PowerChina"])}) + 新矿集团 ({len([n for n in evaluated if n["source"]=="新矿集团"])}) | 共 {len(evaluated)} 条
{len(high_value)}
高价值线索
{len(star3)}
三星线索需复核
{len([e for e in evaluated if e["star"]<=1])}
无效线索
一、高价值线索 ({len(high_value)} 条，自动上传)
{"".join(card(e) for e in high_value) if high_value else '
暂无高价值线索
'}
二、三星线索 ({len(star3)} 条，点击上传)
{"".join(card(e) for e in star3) if star3 else '
暂无三星线索
'}
'''
output_dir = os.path.expanduser(today_dir())
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "线索综合分析报告.html")
with open(output_path, "w", encoding="utf-8") as f:
f.write(html)
print(f"\nHTML 报告已生成: {output_path}")
if __name__ == "__main__":
generate_html_report()