admin/Esmart
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update report interface

Browse Source
This commit is contained in:
lychang
2025-10-01 02:03:21 +08:00
parent be1b3e992c
commit 63c162eb3b
5 changed files with 72982 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.idea/MarsCodeWorkspaceAppSettings.xml generated
View File

@@ -1,7 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="com.codeverse.userSettings.MarscodeWorkspaceAppSettingsState">
<option name="chatAppRouterInfo" value="builder/68d8bbf974c6ec0536baf3c2" />
<option name="chatAppRouterInfo" value="builder/68dc0c18e89aa5cda023dcf6" />
<option name="progress" value="1.0" />
</component>
</project>

456
ai.py
View File

@@ -1,104 +1,380 @@
import json
import os
import time
import logging
from typing import Dict, Any, Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from jimeng import VolcEngineAPIClient
def post_request(url, data, headers=None):
if headers is None:
resp = requests.post(url, json=data)
else:
resp = requests.post(url, json=data, headers=headers)
return resp.json()
class AIAgent:
"""AI代理类，整合了各种AI功能"""
def __init__(self, deepseek_api_key: str = "sk-c2f04b4d635b4df0aace0432f3c6d6f4",
max_retries: int = 3, timeout: int = 30):
"""初始化AI助手
Args:
deepseek_api_key: DeepSeek API密钥
max_retries: 最大重试次数
timeout: 请求超时时间（秒）
"""
self.deepseek_api_key = deepseek_api_key
self.timeout = timeout
def chat_deepseek(message, role):
url = "https://api.deepseek.com/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": "Bearer sk-c2f04b4d635b4df0aace0432f3c6d6f4"
}
body = {
"model": "deepseek-chat",
"messages": [
{"role": "system", "content": role},
{"role": "user", "content": message}
],
"stream": False
}
result = post_request(url, body, headers)
if "choices" in result:
return result['choices'][0]['message']['content']
return ""
# 配置日志
self._setup_logging()
# 创建带重试机制的会话
self.session = self._create_session(max_retries)
def get_image_prompt(message):
role = """你是世界一流的AI生图提示词撰写大师, 你精通心理学、人工智能、工业设计/平面设计、传播学、市场学、哲学、美学，你需要根据用户的需求撰写面向豆包/即梦文生图4.0模型的AI生图提示词。
## 遵循框架
1. 核心风格
这是作品的灵魂，决定了整体的视觉基调。
主导风格： 必须明确指定。例如：写实主义、印象派、赛博朋克、浮世绘、极简主义、超现实主义、儿童插画、蒸汽波、水墨风、概念艺术等。
艺术家或作品参考： 可以具体到某位艺术家的风格（如“梵高的笔触与色彩”、“穆夏的装饰性线条”）或某部著名作品的风格。
技术质感： 描述画作的物理质感。例如：油画厚重的笔触、水彩的晕染感、版画的刻痕、数字绘画的光滑、铅笔素描的颗粒感。
2. 主题与场景
这是作品的叙事发生地，构建了基本语境。
核心主题： 用一句话概括画作讲述的故事或表达的核心概念（如：“孤独的宇航员在异星探险”）。
具体场景： 详细描述环境。例如：繁华的未来都市街角、静谧的森林深处、暴风雨中的海岸、充满温馨灯光的室内书房。
氛围与情感： 定义画面传递的情绪。例如：神秘、宁静、悲壮、欢快、压抑、充满希望。（关键补充） 此部分应指导色彩和构图的选择。
3. 主体内容
这是画面的视觉焦点，最吸引观众视线的部分。
主体对象： 详细描述核心人物、物体或文字。
形态与姿态： 如果是人物或生物，描述其动作、表情、服装细节。
材质与质感： 描述物体的表面特性（如：金属的反光、毛绒的柔软、石头的粗糙、玻璃的透明）。
视角与焦点： 描述观看主体的角度（如：仰视以显雄伟、俯视以显渺小、特写以显细节）。景深是清晰还是模糊。
（关键补充）视觉权重： 说明主体在画面中的大小和位置，以确保其核心地位。
4. 背景与构图
这是支撑主体、深化主题的舞台，决定了画面的空间感和结构。
构图法则： 明确采用的构图方式。例如：三分法、对称式、对角线式、中心式、框架式构图。
背景内容： 描述主体之后的所有元素。例如：远山、云彩、建筑、人群、抽象的光影或色块。背景应与主题和氛围一致。
空间层次： 明确前景、中景、背景的区分，以创造纵深感。例如：前景是几支模糊的芦苇（前景），中景是小船和人物（主体），背景是朦胧的山脉（背景）。
视觉引导： 画面中的线条、光影或色彩如何自然地引导观众的视线走向主体。
5. 色彩与光影
这是渲染氛围、表达情感的关键工具。
色彩基调： 定义主色调和配色方案。例如：暖色调（温馨）、冷色调（清冷）、互补色（强烈）、类比色（和谐）、单色调（高级）。
光影设计：
光源： 描述光的方向（侧光、顶光、逆光）、类型（自然光、烛光、霓虹光）和强度（强烈对比的硬光、柔和的漫射光）。
阴影： 阴影的形状、硬度和透明度，对塑造形体至关重要。
6. 标题与文字集成（如适用）
如果画作包含文字，需进行一体化设计。
文字内容： 明确主标题、副标题或说明文字的具体文案。
字体风格： 选择与绘画风格匹配的字体。例如：衬线体（古典、正式）、无衬线体（现代、简洁）、书法体（优雅、艺术）、手写体（亲切、随意）。
排版与集成： 文字在画面中的位置（顶部、底部、融入背景等）、颜色、大小和透明度，确保它与图像和谐共存，而非生硬粘贴。
7. 整体优化建议（新增项）
一致性检查： 确保风格、场景、色彩、光影等所有元素共同服务于同一个主题和情感。
独特点： 思考画面中最具创新性或最吸引人的“记忆点”是什么。
最终效果： 期望观众看到这幅画时产生怎样的感受或思考。
# 配置常量
self.BASE_REPORT_URL = "https://a.cmdp.cn/basiceg/v1/json/tpldev"
self.CACHE_DIR = "cache"
## 工作流程：
1.分析我给的绘画主题。
2.按照内容要求写出完整详细的绘画设计文本，字数不超过500字
3.按照内容要求逐项填写绘画设计文本
"""
message = f"我的绘画主题是：{message}"
prompt = chat_deepseek(message, role)
return prompt
def _setup_logging(self):
"""配置日志"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
@staticmethod
def _create_session( max_retries: int) -> requests.Session:
"""创建带重试机制的会话
Args:
max_retries: 最大重试次数
Returns:
配置好的会话对象
"""
session = requests.Session()
def get_report_template(security_code,year):
req = requests.get(f"https://a.cmdp.cn/basiceg/v1/json/tpldev/disclosure_assessment_report_prompt/{year}/{security_code}.txt")
if req.status_code != 200:
# 配置重试策略
retry_strategy = Retry(
total=max_retries,
status_forcelist=[429, 500, 502, 503, 504],
method_whitelist=["HEAD", "GET", "POST", "PUT", "DELETE", "OPTIONS", "TRACE"],
backoff_factor=1
)
# 为HTTP和HTTPS适配器添加重试机制
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def _make_request(self, method: str, url: str, **kwargs) -> Optional[Dict[str, Any]]:
"""发送HTTP请求（带错误处理）
Args:
method: HTTP方法
url: 请求URL
**kwargs: 其他请求参数
Returns:
JSON响应或None
"""
try:
# 设置默认超时
if 'timeout' not in kwargs:
kwargs['timeout'] = self.timeout
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
self.logger.info(f"请求成功: {url}")
return response.json()
except requests.exceptions.RequestException as e:
self.logger.error(f"请求失败: {url} - {str(e)}")
return None
except json.JSONDecodeError as e:
self.logger.error(f"JSON解析失败: {url} - {str(e)}")
return None
def chat_deepseek(self, message: str, role: str = "", model: str = "deepseek-chat") -> str:
"""与DeepSeek聊天
Args:
message: 用户消息
role: 系统角色
model: 使用的模型
Returns:
AI回复内容
"""
url = "https://api.deepseek.com/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.deepseek_api_key}"
}
messages = []
if role:
messages.append({"role": "system", "content": role})
messages.append({"role": "user", "content": message})
body = {
"model": model,
"messages": messages,
"stream": False
}
result = self._make_request("POST", url, json=body, headers=headers)
if result and "choices" in result:
return result['choices'][0]['message']['content']
self.logger.warning("DeepSeek API调用失败或返回空结果")
return ""
data = req.text
return data
def generate_report_cache_data(self, year: str) -> Dict[str, Any]:
"""生成报告缓存数据
Args:
year: 年份
Returns:
报告数据
"""
self.logger.info(f"开始生成{year}年报告缓存数据")
base_url = f"{self.BASE_REPORT_URL}/disclosure_assessment_report_data/{year}/"
urls = {
"rank_a": base_url + "Rank_A.json",
"rank_summary": base_url + "Rank_summary.json",
"stats": base_url + "stats.json"
}
# 并行获取数据
data_sources = {}
for key, url in urls.items():
self.logger.info(f"获取{key}数据: {url}")
data = self._make_request("GET", url)
if data:
data_sources[key] = data
else:
self.logger.error(f"获取{key}数据失败")
return {}
# 处理数据
rank_a_data = data_sources["rank_a"]
rank_summary_data = data_sources["rank_summary"]
stats_data = data_sources["stats"]
# 构建数据结构
data = {}
for item in rank_a_data:
data[item["gsdm"]] = {
"name": item["gsjc"],
"num": item["num"],
"ranks": {}
}
# 填充排名数据
for item in rank_summary_data:
security_code = item["gsdm"]
if security_code in data:
year_key = item["kpnd"]
rank = item["kpjg"]
data[security_code]["ranks"][year_key] = rank
# 添加统计信息
data["stats"] = stats_data
# 保存到缓存
self._save_cache(year, data)
self.logger.info(f"{year}年报告缓存数据生成完成")
return data
def _save_cache(self, year: str, data: Dict[str, Any]) -> bool:
"""保存数据到缓存
Args:
year: 年份
data: 要缓存的数据
Returns:
是否保存成功
"""
try:
os.makedirs(self.CACHE_DIR, exist_ok=True)
cache_file = os.path.join(self.CACHE_DIR, f"{year}.json")
with open(cache_file, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
self.logger.info(f"缓存已保存: {cache_file}")
return True
except Exception as e:
self.logger.error(f"保存缓存失败: {str(e)}")
return False
def _load_cache(self, year: str) -> Optional[Dict[str, Any]]:
"""从缓存加载数据
Args:
year: 年份
Returns:
缓存数据或None
"""
cache_file = os.path.join(self.CACHE_DIR, f"{year}.json")
if not os.path.exists(cache_file):
return None
try:
with open(cache_file, "r", encoding="utf-8") as f:
data = json.load(f)
self.logger.info(f"缓存已加载: {cache_file}")
return data
except Exception as e:
self.logger.error(f"加载缓存失败: {str(e)}")
return None
def get_report_data(self, security_code: str, year: str) -> Dict[str, Any]:
"""获取报告数据
Args:
security_code: 证券代码
year: 年份
Returns:
报告数据
"""
# 尝试从缓存加载
cached_data = self._load_cache(year)
if cached_data is None:
# 缓存不存在，生成新数据
cached_data = self.generate_report_cache_data(year)
# 返回特定证券的数据
result = cached_data.get(security_code, {})
result["stats"] = cached_data.get("stats", {})
return result
def get_image_prompt(self, message: str, security_code: str, year: str) -> str:
"""获取图像生成提示词
Args:
message: 绘画主题
security_code: 证券代码
year: 年份
Returns:
图像生成提示词
"""
self.logger.info(f"生成图像提示词: {message}")
with open("cache/painting.prompt", "r", encoding="utf-8") as f:
role = f.read()
# 从缓存中获取特定证券的数据
security_data = self.get_report_data(security_code, year)
security_name = security_data.get("name")
rank = security_data.get("ranks",{}).get(year, "N/A")
num = security_data.get("num", 0)
stats = json.dumps(security_data.get("stats", {}), ensure_ascii=False)
prompt_message = f"## 相关信息：证券名称是{security_name}，{year}年排名是{rank}，已经连续{num}年考评A级。\n{stats}\n\n## 绘画主题：{message}。"
prompt = self.chat_deepseek(prompt_message, role)
if prompt:
self.logger.info("图像提示词生成成功")
else:
self.logger.warning("图像提示词生成失败")
return prompt
def get_report_template(self, security_code: str, year: str) -> str:
"""获取报告模板
Args:
security_code: 证券代码
year: 年份
Returns:
报告模板内容
"""
url = f"{self.BASE_REPORT_URL}/disclosure_assessment_report_prompt/{year}/{security_code}.txt"
try:
response = self.session.get(url, timeout=self.timeout)
response.raise_for_status()
self.logger.info(f"报告模板获取成功: {security_code} - {year}")
return response.text
except requests.exceptions.RequestException as e:
self.logger.error(f"获取报告模板失败: {security_code} - {year} - {str(e)}")
return ""
def get_report(self, security_code: str, year: str) -> str:
"""获取报告
Args:
security_code: 证券代码
year: 年份
Returns:
生成的报告内容
"""
self.logger.info(f"开始生成报告: {security_code} - {year}")
template = self.get_report_template(security_code, year)
if not template:
self.logger.warning("报告模板为空")
return ""
result = self.chat_deepseek(template, "")
if result:
self.logger.info("报告生成成功")
else:
self.logger.warning("报告生成失败")
return result
def get_report(security_code,year):
template = get_report_template(security_code,year)
if not template:
return ""
result = chat_deepseek(template, "")
return result
def test():
"""测试函数"""
# 创建AI助手实例，可配置重试次数和超时时间
assistant = AIAgent(max_retries=3, timeout=30)
try:
# 测试获取报告数据
print("测试获取报告数据...")
data = assistant.get_report_data("000001", "2023")
print(f"获取到的数据: {data}")
# 测试AI聊天
print("\n测试AI聊天...")
response = assistant.chat_deepseek("你好，介绍一下你自己", "你是一个有用的助手")
print(f"AI回复: {response}")
# 测试图像提示词生成
print("\n测试图像提示词生成...")
prompt = assistant.get_image_prompt("春天的花园")
print(f"图像提示词: {prompt}")
# 测试报告生成
print("\n测试报告生成...")
report = assistant.get_report("000001", "2023")
print(f"报告内容: {report}")
except Exception as e:
print(f"测试过程中发生错误: {str(e)}")
assistant.logger.error(f"主函数执行错误: {str(e)}")
agent = AIAgent(max_retries=1, timeout=180)

72547
cache/2023.json vendored Normal file
View File

File diff suppressed because it is too large Load Diff

49
cache/painting.prompt vendored Normal file
View File

@@ -0,0 +1,49 @@
你是世界一流的AI生图提示词撰写大师, 你精通心理学、人工智能、工业设计/平面设计、传播学、市场学、哲学、美学，你需要根据用户的需求撰写面向豆包/即梦文生图4.0模型的AI生图提示词。
## 遵循框架
1. 核心风格
风格必须基于绘画主题确定，不允许主题文字出现在图片上。
主导风格：根据主题语境指定，例如：若主题为“孤独探险”，可选超现实主义或写实主义；若主题为“未来城市”，可选赛博朋克或概念艺术。
艺术家或作品参考：如主题需情感张力，参考梵高笔触；需装饰性，参考穆夏线条。
技术质感：匹配主题氛围，如油画厚重感用于古典主题，数字光滑感用于科技主题。
2. 主题与场景
主题不显示为文字，仅通过视觉元素表达。
核心主题：用一句话概括叙事，如“一位旅者在暮色森林中寻找归宿”。
具体场景：描述环境细节，如幽暗的丛林小径、霓虹闪烁的都市小巷，以强化主题。
氛围与情感：定义情绪（如孤寂、希望），指导色彩与构图选择。
3. 主体内容
聚焦视觉焦点，确保主体突出。
主体对象：详细描述核心元素，如人物、物体或文字（文字内容必须来自用户提供的“相关信息”，确保准确无误）。
材质与质感：表面特性如金属反光、织物柔软。
视角与焦点：采用仰视、特写等角度，景深清晰或模糊以突出主体。
视觉权重：主体大小占画面主导位置。
4. 背景与构图
支撑主体，深化主题叙事。
构图法则：如三分法、对称式，确保视觉平衡。
背景内容：元素如远山、建筑，与主题一致；文字可融入背景（位置、颜色基于“相关信息”指定）。
空间层次：区分前景、中景、背景以增强纵深感。
视觉引导：用线条、光影引导视线至主体或文字。
5. 色彩与光影
渲染氛围，传递情感。
色彩基调：主色调和配色方案，如冷色调表达清冷，暖色调表达温馨。
光影设计：光源方向（如侧光）、类型（如自然光）、强度（如柔和光）；阴影形状与透明度以塑形。
6. 标题与文字集成（如适用）
强制使用用户“相关信息”中的文本内容，确保文字准确、完整。
文字内容：直接引用提供的文案，如主标题、副标题。
字体风格：匹配绘画风格，如无衬线体用于现代主题。
排版与集成：文字位置（如底部、融入背景）、颜色、大小和透明度，与图像和谐统一。
7. 整体优化建议
一致性检查：所有元素服务于同一主题和情感。
独特点：突出创新元素，如独特光影或文字设计。
最终效果：目标观众感受，如引发沉思或共鸣。
## 工作流程
- 风格由主题隐性确定，绝不显示主题文字于图片。
- 图片中出现的文本内容严格基于用户“相关信息”，确保准确性。
- 直接生成提示词，字数不超过500字。

31
index.py
View File

@@ -4,7 +4,7 @@ import base64
import re
import requests
from ai import get_image_prompt, get_report, get_report_template
from ai import agent
from jimeng import VolcEngineAPIClient
from router import Router, Response
from esmart import transform_data, SmartSheet, ESmart, push_message
@@ -218,17 +218,23 @@ def image_generate(request):
prompt = request.body.get('prompt', "")
source = request.body.get('source', "url")
user = request.body.get('user', {})
if prompt:
prompt = get_image_prompt(prompt)
result = vea.run(prompt, source, 1)
if source != "url":
result = str(base64.b64encode(result))[2:-1]
return Response(result)
else:
security_code = user.get('seccode', "")
year = user.get('year', "2023")
if not security_code:
return Response({
'success': False,
'message': f'user.seccode 不能为空'
}, 400)
if not prompt:
return Response({
'success': False,
'message': f'prompt 不能为空'
}, 400)
prompt = agent.get_image_prompt(prompt,security_code,year)
result = vea.run(prompt, source, 1)
if source != "url":
result = str(base64.b64encode(result))[2:-1]
return Response(result)
@router.route('/ai/image/prompt', methods=['POST'])
@@ -237,7 +243,7 @@ def image_prompt(request):
"""获取绘制图像详细提示词"""
prompt = request.body.get('prompt', "")
if prompt:
prompt = get_image_prompt(prompt)
prompt = agent.get_image_prompt(prompt)
return Response(prompt)
else:
return Response({
@@ -256,7 +262,7 @@ def report_generate(request):
'success': False,
'message': f'seccode 不能为空'
})
result = get_report(security_code,year)
result = agent.get_report(security_code, year)
if result:
return Response(result)
else:
@@ -268,7 +274,7 @@ def report_generate(request):
@router.route('/ai/report/prompt', methods=['POST'])
def report_prompt(request):
"""生成相关报告"""
"""查询相关报告的提示词"""
security_code = request.body.get('seccode', "")
year = request.body.get('year', "2023")
if not security_code:
@@ -276,7 +282,7 @@ def report_prompt(request):
'success': False,
'message': f'seccode 不能为空'
})
result = get_report_template(security_code,year)
result = agent.get_report_template(security_code, year)
if result:
return Response(result)
else:
@@ -285,5 +291,6 @@ def report_prompt(request):
'message': f'无相关报告'
}, 400)
def main_handler(event, context):
return router.handle_request(event)