|
|
@@ -0,0 +1,293 @@
|
|
|
+import json
|
|
|
+from collections import defaultdict
|
|
|
+from pathlib import Path
|
|
|
+
|
|
|
+from examples.demand.data_query_tools import get_changwen_weight, get_zengzhang_weight
|
|
|
+from examples.demand.db_manager import DatabaseManager
|
|
|
+from examples.demand.models import TopicPatternElement, TopicPatternExecution
|
|
|
+from examples.demand.pattern_builds.pattern_service import run_mining
|
|
|
+
|
|
|
+db = DatabaseManager()
|
|
|
+
|
|
|
+ZENGZHANG_DATA_DIR = Path(__file__).parent / "data" / "zengzhang_data"
|
|
|
+
|
|
|
+
|
|
|
+def _safe_float(value):
|
|
|
+ if value is None:
|
|
|
+ return 0.0
|
|
|
+ try:
|
|
|
+ return float(value)
|
|
|
+ except (TypeError, ValueError):
|
|
|
+ return 0.0
|
|
|
+
|
|
|
+
|
|
|
+def _build_category_scores(name_scores, name_paths, name_post_ids):
|
|
|
+ node_scores = defaultdict(float)
|
|
|
+ node_post_ids = defaultdict(set)
|
|
|
+ for name, score in name_scores.items():
|
|
|
+ paths = name_paths.get(name, set())
|
|
|
+ post_ids = name_post_ids.get(name, set())
|
|
|
+ for category_path in paths:
|
|
|
+ if not category_path:
|
|
|
+ continue
|
|
|
+ nodes = [segment.strip() for segment in category_path.split(">") if segment.strip()]
|
|
|
+ for idx in range(len(nodes)):
|
|
|
+ prefix = ">".join(nodes[: idx + 1])
|
|
|
+ node_scores[prefix] += score
|
|
|
+ if post_ids:
|
|
|
+ node_post_ids[prefix].update(post_ids)
|
|
|
+ return node_scores, node_post_ids
|
|
|
+
|
|
|
+
|
|
|
+def _write_json(path, payload):
|
|
|
+ with open(path, "w", encoding="utf-8") as f:
|
|
|
+ json.dump(payload, f, ensure_ascii=False, indent=2)
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_post_id(value):
|
|
|
+ """
|
|
|
+ 统一 post_id/videoid 格式，降低源数据格式差异导致的匹配失败。
|
|
|
+ """
|
|
|
+ if value is None:
|
|
|
+ return ""
|
|
|
+ s = str(value).strip()
|
|
|
+ if not s:
|
|
|
+ return ""
|
|
|
+ if s.endswith(".0"):
|
|
|
+ s = s[:-2]
|
|
|
+ return s
|
|
|
+
|
|
|
+
|
|
|
+def _extract_digits(value: str) -> str:
|
|
|
+ if not value:
|
|
|
+ return ""
|
|
|
+ return "".join(ch for ch in value if ch.isdigit())
|
|
|
+
|
|
|
+
|
|
|
+def _build_score_by_videoid(cluster_name: str):
|
|
|
+ json_path = ZENGZHANG_DATA_DIR / f"{cluster_name}.json"
|
|
|
+ with open(json_path, "r", encoding="utf-8") as f:
|
|
|
+ payload = json.load(f)
|
|
|
+
|
|
|
+ if not isinstance(payload, list):
|
|
|
+ raise ValueError(f"数据格式错误，期望数组: {json_path}")
|
|
|
+
|
|
|
+ score_map = {}
|
|
|
+ for item in payload:
|
|
|
+ if not isinstance(item, dict):
|
|
|
+ continue
|
|
|
+ videoid = item.get("videoid")
|
|
|
+ if videoid is None:
|
|
|
+ continue
|
|
|
+ ext_data = item.get("ext_data") or {}
|
|
|
+ if not isinstance(ext_data, dict):
|
|
|
+ continue
|
|
|
+ realplay_uv = _safe_float(ext_data.get("分发realplay_uv"))
|
|
|
+ exposure_uv = _safe_float(ext_data.get("当日分发曝光uv"))
|
|
|
+ norm_videoid = _normalize_post_id(videoid)
|
|
|
+ if not norm_videoid:
|
|
|
+ continue
|
|
|
+ score_map[norm_videoid] = (realplay_uv / exposure_uv) if exposure_uv > 0 else 0.0
|
|
|
+ return score_map
|
|
|
+
|
|
|
+
|
|
|
+def filter_low_exposure_records(
|
|
|
+ cluster_name: str = None,
|
|
|
+ min_exposure: float = 1000,
|
|
|
+):
|
|
|
+ """
|
|
|
+ 过滤 JSON 中推荐曝光数小于阈值的记录，并写回原文件。
|
|
|
+ 默认过滤阈值: 1000
|
|
|
+ """
|
|
|
+ json_path = ZENGZHANG_DATA_DIR / f"{cluster_name}.json"
|
|
|
+ with open(json_path, "r", encoding="utf-8") as f:
|
|
|
+ payload = json.load(f)
|
|
|
+
|
|
|
+ if not isinstance(payload, list):
|
|
|
+ raise ValueError(f"数据格式错误，期望数组: {json_path}")
|
|
|
+
|
|
|
+ filtered = []
|
|
|
+ for item in payload:
|
|
|
+ if not isinstance(item, dict):
|
|
|
+ continue
|
|
|
+ ext_data = item.get("ext_data") or {}
|
|
|
+ exposure = _safe_float(ext_data.get("推荐曝光数")) if isinstance(ext_data, dict) else 0.0
|
|
|
+ if exposure >= float(min_exposure):
|
|
|
+ filtered.append(item)
|
|
|
+
|
|
|
+ with open(json_path, "w", encoding="utf-8") as f:
|
|
|
+ json.dump(filtered, f, ensure_ascii=False, indent=2)
|
|
|
+
|
|
|
+ return {
|
|
|
+ "file": str(json_path),
|
|
|
+ "before_count": len(payload),
|
|
|
+ "after_count": len(filtered),
|
|
|
+ "removed_count": len(payload) - len(filtered),
|
|
|
+ "min_exposure": float(min_exposure),
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def zengzhang_data_prepare(cluster_name) -> int:
|
|
|
+ json_path = ZENGZHANG_DATA_DIR / f"{cluster_name}.json"
|
|
|
+ with open(json_path, "r", encoding="utf-8") as f:
|
|
|
+ payload = json.load(f)
|
|
|
+
|
|
|
+ if not isinstance(payload, list):
|
|
|
+ raise ValueError(f"数据格式错误，期望数组: {json_path}")
|
|
|
+
|
|
|
+ video_ids = []
|
|
|
+ for item in payload:
|
|
|
+ if not isinstance(item, dict):
|
|
|
+ continue
|
|
|
+ video_id = item.get("videoid")
|
|
|
+ if video_id is None:
|
|
|
+ continue
|
|
|
+ video_id_str = str(video_id).strip()
|
|
|
+ if video_id_str:
|
|
|
+ video_ids.append(video_id_str)
|
|
|
+
|
|
|
+ # 去重并保持原有顺序，避免重复挖掘同一视频
|
|
|
+ video_ids = list(dict.fromkeys(video_ids))
|
|
|
+ if not video_ids:
|
|
|
+ raise ValueError(f"未在文件中解析到有效 videoid: {json_path}")
|
|
|
+
|
|
|
+ execution_id = run_mining(post_ids=video_ids, cluster_name=cluster_name, platform='zengzhang')
|
|
|
+ return execution_id
|
|
|
+
|
|
|
+
|
|
|
+def prepare_by_json_score(execution_id: int, cluster_name: str = ""):
|
|
|
+ """
|
|
|
+ 与 prepare.py 的输出结构保持一致，但分数来源改为:
|
|
|
+ score = score = 分发realplay_uv / 当日分发曝光uv
|
|
|
+ """
|
|
|
+ session = db.get_session()
|
|
|
+ try:
|
|
|
+ execution = session.query(TopicPatternExecution).filter(
|
|
|
+ TopicPatternExecution.id == execution_id
|
|
|
+ ).first()
|
|
|
+ if not execution:
|
|
|
+ raise ValueError(f"execution_id 不存在: {execution_id}")
|
|
|
+
|
|
|
+ score_by_post_id = _build_score_by_videoid(cluster_name)
|
|
|
+ rows = session.query(TopicPatternElement).filter(
|
|
|
+ TopicPatternElement.execution_id == execution_id
|
|
|
+ ).all()
|
|
|
+ if not rows:
|
|
|
+ return {"message": "没有可处理的数据", "execution_id": execution_id}
|
|
|
+
|
|
|
+ grouped = {
|
|
|
+ "实质": {"name_post_ids": defaultdict(set), "name_paths": defaultdict(set)},
|
|
|
+ "形式": {"name_post_ids": defaultdict(set), "name_paths": defaultdict(set)},
|
|
|
+ "意图": {"name_post_ids": defaultdict(set), "name_paths": defaultdict(set)},
|
|
|
+ }
|
|
|
+
|
|
|
+ for r in rows:
|
|
|
+ element_type = (r.element_type or "").strip()
|
|
|
+ if element_type not in grouped:
|
|
|
+ continue
|
|
|
+ name = (r.name or "").strip()
|
|
|
+ if not name:
|
|
|
+ continue
|
|
|
+ if r.post_id:
|
|
|
+ grouped[element_type]["name_post_ids"][name].add(str(r.post_id))
|
|
|
+ if r.category_path:
|
|
|
+ grouped[element_type]["name_paths"][name].add(r.category_path.strip())
|
|
|
+
|
|
|
+ output_dir = Path(__file__).parent / "data" / str(execution_id)
|
|
|
+ output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
+ match_stats = {
|
|
|
+ "post_ids_total": 0,
|
|
|
+ "post_ids_scored_direct": 0,
|
|
|
+ "post_ids_scored_by_digits": 0,
|
|
|
+ "post_ids_missing_score": 0,
|
|
|
+ }
|
|
|
+ summary = {
|
|
|
+ "execution_id": execution_id,
|
|
|
+ "merge_leve2": execution.merge_leve2,
|
|
|
+ "files": {},
|
|
|
+ "score_match_stats": match_stats,
|
|
|
+ }
|
|
|
+
|
|
|
+ for element_type, data in grouped.items():
|
|
|
+ name_post_ids = data["name_post_ids"]
|
|
|
+ name_paths = data["name_paths"]
|
|
|
+
|
|
|
+ name_scores = {}
|
|
|
+ for name, post_ids in name_post_ids.items():
|
|
|
+ scores = []
|
|
|
+ for raw_pid in post_ids:
|
|
|
+ match_stats["post_ids_total"] += 1
|
|
|
+ pid = _normalize_post_id(raw_pid)
|
|
|
+ score = score_by_post_id.get(pid)
|
|
|
+ if score is not None:
|
|
|
+ match_stats["post_ids_scored_direct"] += 1
|
|
|
+ scores.append(_safe_float(score))
|
|
|
+ continue
|
|
|
+
|
|
|
+ # 兜底：当 post_id 含前后缀时，尝试仅用数字部分匹配 videoid
|
|
|
+ digits_pid = _extract_digits(pid)
|
|
|
+ if digits_pid and digits_pid in score_by_post_id:
|
|
|
+ match_stats["post_ids_scored_by_digits"] += 1
|
|
|
+ scores.append(_safe_float(score_by_post_id[digits_pid]))
|
|
|
+ else:
|
|
|
+ match_stats["post_ids_missing_score"] += 1
|
|
|
+ scores.append(0.0)
|
|
|
+ name_scores[name] = (sum(scores) / len(scores)) if scores else 0.0
|
|
|
+
|
|
|
+ raw_elements = []
|
|
|
+ for name, score in name_scores.items():
|
|
|
+ post_ids_set = name_post_ids.get(name, set())
|
|
|
+ raw_elements.append(
|
|
|
+ {
|
|
|
+ "name": name,
|
|
|
+ "score": round(score, 6),
|
|
|
+ "post_ids_count": len(post_ids_set),
|
|
|
+ "category_paths": sorted(list(name_paths.get(name, set()))),
|
|
|
+ }
|
|
|
+ )
|
|
|
+
|
|
|
+ element_payload = sorted(raw_elements, key=lambda x: (-x["score"], x["name"]))
|
|
|
+ category_scores, category_post_ids = _build_category_scores(
|
|
|
+ name_scores, name_paths, name_post_ids
|
|
|
+ )
|
|
|
+ category_payload = sorted(
|
|
|
+ [
|
|
|
+ {
|
|
|
+ "category_path": path,
|
|
|
+ "category": path.split(">")[-1].strip() if path else "",
|
|
|
+ "score": round(score, 6),
|
|
|
+ "post_ids_count": len(category_post_ids.get(path, set())),
|
|
|
+ }
|
|
|
+ for path, score in category_scores.items()
|
|
|
+ ],
|
|
|
+ key=lambda x: x["score"],
|
|
|
+ reverse=True,
|
|
|
+ )
|
|
|
+
|
|
|
+ element_file = output_dir / f"{element_type}_元素.json"
|
|
|
+ category_file = output_dir / f"{element_type}_分类.json"
|
|
|
+ _write_json(element_file, element_payload)
|
|
|
+ _write_json(category_file, category_payload)
|
|
|
+ summary["files"][f"{element_type}_元素"] = str(element_file)
|
|
|
+ summary["files"][f"{element_type}_分类"] = str(category_file)
|
|
|
+
|
|
|
+ return summary
|
|
|
+ finally:
|
|
|
+ session.close()
|
|
|
+
|
|
|
+
|
|
|
+def zengzhang_prepare(cluster_name):
|
|
|
+ get_zengzhang_weight(cluster_name)
|
|
|
+ filter_low_exposure_records(cluster_name=cluster_name)
|
|
|
+ execution_id = zengzhang_data_prepare(cluster_name)
|
|
|
+ if execution_id:
|
|
|
+ print(f"execution_id={execution_id}")
|
|
|
+ print(prepare_by_json_score(execution_id, cluster_name))
|
|
|
+ return execution_id
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ # 奇观妙技有乾坤 青史铁事漫谈
|
|
|
+ cluster_name = '新余'
|
|
|
+ execution_id = zengzhang_prepare(cluster_name=cluster_name)
|
|
|
+ print(execution_id)
|
xueyiming