File size: 6,623 Bytes
f71610e |
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 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 |
import gradio as gr
import pandas as pd
from sqlalchemy import create_engine, text
# 数据库连接配置
DB_CONFIG = {
'host': 'rm-j6c5yhe0l739e7752vo.mysql.cnhk.rds.aliyuncs.com',
'user': 'report_user',
'password': 'report_user_123',
'database': 'easy_financial_report'
}
def get_database_url():
"""构造数据库连接URL"""
return f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}/{DB_CONFIG['database']}"
def execute_query_with_connection(engine, query, params=None):
"""
自动打开和关闭数据库连接执行查询
Args:
engine: SQLAlchemy引擎实例
query: SQL查询语句
params: 查询参数(可选)
Returns:
查询结果或错误信息
"""
try:
with engine.connect() as conn:
if params:
result = conn.execute(text(query), params)
else:
result = conn.execute(text(query))
return result.fetchall()
except Exception as e:
return f"查询执行失败: {str(e)}"
def execute_query(query):
"""执行SQL查询并返回结果"""
if not query.strip():
return "请输入SQL查询语句"
try:
# 创建数据库连接引擎
engine = create_engine(get_database_url())
# 使用上下文管理器执行查询
with engine.connect() as conn:
df = pd.read_sql_query(text(query), conn)
return df
except Exception as e:
return f"查询执行失败: {str(e)}"
# 引擎会在with语句结束后自动清理连接
def get_table_names():
"""获取数据库中的所有表名"""
try:
# 创建数据库连接引擎
engine = create_engine(get_database_url())
# 使用上下文管理器执行查询
with engine.connect() as conn:
# 查询所有表名
query = "SHOW TABLES"
df = pd.read_sql_query(text(query), conn)
# 返回表名列表
return df.iloc[:, 0].tolist() if not df.empty else []
except Exception as e:
return [f"获取表名失败: {str(e)}"]
# 引擎会在with语句结束后自动清理连接
def preview_table(table_name):
"""预览表的前几行数据"""
if not table_name or "失败" in table_name:
return "请选择有效的表名"
query = f"SELECT * FROM {table_name} LIMIT 10"
return execute_query(query)
# 新增功能函数
def insert_record(title):
"""向report_file_link表插入新记录"""
if not title.strip():
return "请输入标题"
try:
engine = create_engine(get_database_url())
# 插入新记录
query = "INSERT INTO report_file_link (title) VALUES (:title)"
with engine.connect() as conn:
trans = conn.begin()
try:
conn.execute(text(query), {"title": title})
trans.commit()
return f"成功插入记录: {title}"
except Exception as e:
trans.rollback()
raise e
except Exception as e:
return f"插入记录失败: {str(e)}"
# 引擎会在with语句结束后自动清理连接
def update_record(record_id, new_title):
"""更新report_file_link表中的记录"""
if not record_id or not new_title.strip():
return "请输入记录ID和新标题"
try:
engine = create_engine(get_database_url())
# 更新记录
query = "UPDATE report_file_link SET title = :title WHERE id = :id"
with engine.connect() as conn:
trans = conn.begin()
try:
result = conn.execute(text(query), {"title": new_title, "id": record_id})
trans.commit()
if result.rowcount > 0:
return f"成功更新记录ID {record_id} 的标题为: {new_title}"
else:
return f"未找到ID为 {record_id} 的记录"
except Exception as e:
trans.rollback()
raise e
except Exception as e:
return f"更新记录失败: {str(e)}"
# 引擎会在with语句结束后自动清理连接
def delete_record(record_id):
"""从report_file_link表中删除记录"""
if not record_id:
return "请输入记录ID"
try:
engine = create_engine(get_database_url())
# 删除记录
query = "DELETE FROM report_file_link WHERE id = :id"
with engine.connect() as conn:
trans = conn.begin()
try:
result = conn.execute(text(query), {"id": record_id})
trans.commit()
if result.rowcount > 0:
return f"成功删除ID为 {record_id} 的记录"
else:
return f"未找到ID为 {record_id} 的记录"
except Exception as e:
trans.rollback()
raise e
except Exception as e:
return f"删除记录失败: {str(e)}"
# 引擎会在with语句结束后自动清理连接
def refresh_report_file_link():
"""刷新report_file_link表的数据"""
return execute_query("SELECT * FROM report_file_link")
# 新增功能函数
def insert_company(company_name, stock_code):
try:
engine = create_engine(get_database_url())
# 插入新记录
query = "INSERT INTO company (company_name, stock_code) VALUES (:company_name, :stock_code)"
with engine.connect() as conn:
trans = conn.begin()
try:
conn.execute(text(query), {"company_name": company_name, "stock_code": stock_code})
trans.commit()
return True
except Exception as e:
trans.rollback()
raise e
except Exception as e:
return False
# 引擎会在with语句结束后自动清理连接
def get_companys():
"""获取company表中的所有公司"""
query = "SELECT * FROM company"
return execute_query(query)
def get_company_by_name(company_name):
"""根据公司名称获取公司信息"""
query = "SELECT * FROM company WHERE company_name = :company_name"
try:
engine = create_engine(get_database_url())
with engine.connect() as conn:
df = pd.read_sql_query(text(query), conn, params={"company_name": company_name})
return df
except Exception as e:
return f"查询执行失败: {str(e)}" |