{"contents":"# -*- coding: utf-8 -*-#\n# -------------------------------------------------------------------------------\n# Name:         analytics.py\n# Description:  统计分析（wordcloud 等）\n# -------------------------------------------------------------------------------\nfrom collections import Counter\n\nfrom pydantic import BaseModel\nfrom fastapi import APIRouter\n\nfrom wxdump_linux.db import DBHandler\nfrom ..response import ReJson\nfrom ..decorators import error9999\nfrom ..config import gc\n\nanalytics_router = APIRouter()\n\n\nclass DateCountRequest(BaseModel):\n    wxid: str = \"\"\n    start_time: int = 0\n    end_time: int = 0\n    time_format: str = \"%Y-%m-%d\"\n\n\n@analytics_router.api_route('/date_count', methods=[\"GET\", 'POST'])\ndef get_date_count(request: DateCountRequest):\n    \"\"\"\n    获取日期统计\n    :return:\n    \"\"\"\n    wxid = request.wxid\n    start_time = request.start_time\n    end_time = request.end_time\n    time_format = request.time_format\n\n    my_wxid = gc.get_conf(gc.at, \"last\")\n    if not my_wxid: return ReJson(1001, body=\"my_wxid is required\")\n    db_config = gc.get_conf(my_wxid, \"db_config\")\n    db = DBHandler(db_config, my_wxid=my_wxid)\n    date_count = db.get_date_count(wxid=wxid, start_time=start_time, end_time=end_time, time_format=time_format)\n    return ReJson(0, date_count)\n\n\nclass TopTalkerCountRequest(BaseModel):\n    top: int = 10\n    start_time: int = 0\n    end_time: int = 0\n\n\n@analytics_router.api_route('/top_talker_count', methods=[\"GET\", 'POST'])\ndef get_top_talker_count(request: TopTalkerCountRequest):\n    \"\"\"\n    获取最多聊天的人\n    :return:\n    \"\"\"\n    top = request.top\n    start_time = request.start_time\n    end_time = request.end_time\n\n    my_wxid = gc.get_conf(gc.at, \"last\")\n    if not my_wxid: return ReJson(1001, body=\"my_wxid is required\")\n    db_config = gc.get_conf(my_wxid, \"db_config\")\n    date_count = DBHandler(db_config, my_wxid=my_wxid).get_top_talker_count(top=top, start_time=start_time,\n                                                                            end_time=end_time)\n    return ReJson(0, date_count)\n\n\nclass WordCloudRequest(BaseModel):\n    target: str = \"signature\"\n\n\n@analytics_router.api_route('/wordcloud', methods=[\"GET\", 'POST'])\n@error9999\ndef get_wordcloud(request: WordCloudRequest):\n    try:\n        import jieba\n    except ImportError:\n        return ReJson(9999, body=\"jieba is required\")\n\n    target = request.target\n    if not target:\n        return ReJson(1002, body=\"target is required\")\n\n    my_wxid = gc.get_conf(gc.at, \"last\")\n    if not my_wxid: return ReJson(1001, body=\"my_wxid is required\")\n    db_config = gc.get_conf(my_wxid, \"db_config\")\n    db = DBHandler(db_config, my_wxid=my_wxid)\n\n    if target == \"signature\":\n        users = db.get_user()\n        signature_list = []\n        for wxid, user in users.items():\n            ExtraBuf = user.get(\"ExtraBuf\", {})\n            signature = ExtraBuf.get(\"个性签名\", \"\") if ExtraBuf else \"\"\n            if signature:\n                signature_list.append(signature)\n        signature_str = \" \".join(signature_list)\n        words = jieba.lcut(signature_str)\n        words = [word for word in words if len(word) \u003e 1]\n        count_dict = dict(Counter(words))\n        return ReJson(0, count_dict)\n    elif target == \"nickname\":\n        users = db.get_user()\n        nickname_list = []\n        for wxid, user in users.items():\n            nickname = user.get(\"nickname\", \"\")\n            if nickname:\n                nickname_list.append(nickname)\n        nickname_str = \" \".join(nickname_list)\n        words = jieba.lcut(nickname_str)\n        words = [word for word in words if len(word) \u003e 1]\n        count_dict = dict(Counter(words))\n        return ReJson(0, count_dict)\n\n    return ReJson(1002, body=\"target is required\")\n","is_binary":false,"path":"wxdump_linux/api/routes/analytics.py","ref":""}