File size: 3,574 Bytes
61517de |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 |
import requests
from config import conf
from common.log import logger
import os
class LinkSummary:
def __init__(self):
pass
def summary_file(self, file_path: str):
file_body = {
"file": open(file_path, "rb"),
"name": file_path.split("/")[-1],
}
url = self.base_url() + "/v1/summary/file"
res = requests.post(url, headers=self.headers(), files=file_body, timeout=(5, 300))
return self._parse_summary_res(res)
def summary_url(self, url: str):
body = {
"url": url
}
res = requests.post(url=self.base_url() + "/v1/summary/url", headers=self.headers(), json=body, timeout=(5, 180))
return self._parse_summary_res(res)
def summary_chat(self, summary_id: str):
body = {
"summary_id": summary_id
}
res = requests.post(url=self.base_url() + "/v1/summary/chat", headers=self.headers(), json=body, timeout=(5, 180))
if res.status_code == 200:
res = res.json()
logger.debug(f"[LinkSum] chat open, res={res}")
if res.get("code") == 200:
data = res.get("data")
return {
"questions": data.get("questions"),
"file_id": data.get("file_id")
}
else:
res_json = res.json()
logger.error(f"[LinkSum] summary error, status_code={res.status_code}, msg={res_json.get('message')}")
return None
def _parse_summary_res(self, res):
if res.status_code == 200:
res = res.json()
logger.debug(f"[LinkSum] url summary, res={res}")
if res.get("code") == 200:
data = res.get("data")
return {
"summary": data.get("summary"),
"summary_id": data.get("summary_id")
}
else:
res_json = res.json()
logger.error(f"[LinkSum] summary error, status_code={res.status_code}, msg={res_json.get('message')}")
return None
def base_url(self):
return conf().get("linkai_api_base", "https://api.link-ai.chat")
def headers(self):
return {"Authorization": "Bearer " + conf().get("linkai_api_key")}
def check_file(self, file_path: str, sum_config: dict) -> bool:
file_size = os.path.getsize(file_path) // 1000
if (sum_config.get("max_file_size") and file_size > sum_config.get("max_file_size")) or file_size > 15000:
logger.warn(f"[LinkSum] file size exceeds limit, No processing, file_size={file_size}KB")
return False
suffix = file_path.split(".")[-1]
support_list = ["txt", "csv", "docx", "pdf", "md", "jpg", "jpeg", "png"]
if suffix not in support_list:
logger.warn(f"[LinkSum] unsupported file, suffix={suffix}, support_list={support_list}")
return False
return True
def check_url(self, url: str):
if not url:
return False
support_list = ["http://mp.weixin.qq.com", "https://mp.weixin.qq.com"]
black_support_list = ["https://mp.weixin.qq.com/mp/waerrpage"]
for black_url_prefix in black_support_list:
if url.strip().startswith(black_url_prefix):
logger.warn(f"[LinkSum] unsupported url, no need to process, url={url}")
return False
for support_url in support_list:
if url.strip().startswith(support_url):
return True
return False
|