pfqgauxfb
/
UAV-system
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

drone_decision_system

Browse Source
main
刘宇杰 2 months ago
parent 8d2137f648
commit 5fafbe2dfd
68 changed files with 2515 additions and 0 deletions
Show Stats Download Patch File Download Diff File

39
drone_decision_system/app.py
Unescape View File

@ -0,0 +1,39 @@
"""
无人机决策系统 - 主程序入口
作者：刘宇杰
日期：2025-07-13
功能说明：
本文件为系统入口，负责初始化配置、加载主界面、启动应用。
"""
import os
from gui.login_window import LoginWindow
from gui.main_window import MainWindow
from config.settings import settings
from utils.file_utils import FileUtils
def main():
"""
应用程序入口点。
负责初始化并启动登录窗口，登录成功后进入主界面。
"""
# 确保必要的目录存在
# 如果uploads是文件，先删除
if os.path.exists(settings.UPLOAD_FOLDER) and os.path.isfile(settings.UPLOAD_FOLDER):
os.remove(settings.UPLOAD_FOLDER)
FileUtils.create_secure_upload_dir(settings.UPLOAD_FOLDER)
def on_login_success(user):
"""
登录成功后的回调函数。
参数：
user (dict): 当前登录的用户信息。
"""
# 打开主窗口
MainWindow(user)
# 显示登录窗口
LoginWindow(on_login_success)
if __name__ == "__main__":
main()

BIN
drone_decision_system/config/__pycache__/settings.cpython-313.pyc
View File

Binary file not shown.

34
drone_decision_system/config/settings.py
Unescape View File

@ -0,0 +1,34 @@
"""
无人机决策系统 - 配置文件
作者：刘宇杰
日期：2025-07-13
功能说明：
本文件定义全局配置参数，包括API密钥、数据库路径、上传目录、允许的文件类型等。
"""
import os
from pathlib import Path
# 基础配置
class Settings:
# 项目根目录
BASE_DIR = Path(__file__).resolve().parent.parent
# 文心大模型API配置
ERNIE_API_KEY = "bce-v3/ALTAK-d9EuFZrO8ZIViw4fywXCp/bd1c6d19fe2a2596613298e292ef38c845064a30" # 替换为你的API密钥
ERNIE_API_URL = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/ERNIE-4.5-Turbo-128K-Preview"
# 数据库配置
DATABASE_PATH = os.path.join(BASE_DIR, "data", "history.db")
# 文件上传配置
UPLOAD_FOLDER = os.path.join(BASE_DIR, "data", "uploads")
ALLOWED_IMAGE_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
ALLOWED_TEXT_EXTENSIONS = {'txt', 'pdf', 'docx'}
# 界面样式配置
PRIMARY_COLOR = "#2c3e50"
SECONDARY_COLOR = "#3498db"
BACKGROUND_COLOR = "#ecf0f1"
TEXT_COLOR = "#2c3e50"
settings = Settings()

BIN
drone_decision_system/core/__pycache__/analysis.cpython-313.pyc
View File

Binary file not shown.

BIN
drone_decision_system/core/__pycache__/ernie_client.cpython-313.pyc
View File

Binary file not shown.

BIN
drone_decision_system/core/__pycache__/image_processor.cpython-313.pyc
View File

Binary file not shown.

BIN
drone_decision_system/core/__pycache__/text_processor.cpython-313.pyc
View File

Binary file not shown.

232
drone_decision_system/core/analysis.py
Unescape View File

@ -0,0 +1,232 @@
"""
无人机决策系统 - 分析引擎模块
作者：刘宇杰
日期：2025-07-13
功能说明：
本模块负责统一调度文本、图像分析，调用ERNIE大模型API，生成分析结果和打击计划。
"""
from typing import Dict, Any, List
from datetime import datetime
from config.settings import settings
from .ernie_client import ErnieClient
from .image_processor import ImageProcessor
from .text_processor import TextProcessor
import json
class AnalysisEngine:
"""
分析引擎类。
负责协调文本和图像分析，处理分析结果，生成无人机打击计划。
"""
def __init__(self):
"""
初始化分析引擎。
创建ERNIE客户端、图像处理器和文本处理器实例。
"""
self.ernie_client = ErnieClient()
self.image_processor = ImageProcessor()
self.text_processor = TextProcessor()
def analyze_text(self, text_content: str, filename: str) -> Dict[str, Any]:
"""
分析文本内容。
参数：
text_content (str): 文本内容。
filename (str): 原始文件名。
返回：
dict: 分析结果。
"""
# 保存文本
text_path = self.text_processor.save_uploaded_text(text_content, filename)
if not text_path:
return {"error": "文本保存失败"}
# 提取关键信息(可选)
key_info = self.text_processor.extract_key_info(text_content)
# 调用文心大模型进行深入分析
ernie_result = self.ernie_client.analyze_text(text_content)
# 合并结果
result = {
"type": "text",
"file_path": text_path,
"key_info": key_info,
"ernie_analysis": self._validate_ernie_result(ernie_result),
"timestamp": datetime.now().isoformat()
}
return result
def analyze_image(self, image_data: bytes, filename: str) -> Dict[str, Any]:
"""
分析图像内容。
参数：
image_data (bytes): 图像二进制数据。
filename (str): 原始文件名。
返回：
dict: 分析结果。
"""
# 保存图像
image_path = self.image_processor.save_uploaded_image(image_data, filename)
if not image_path:
return {"error": "图像保存失败"}
# 预处理图像
processed_image = self.image_processor.process_image_for_analysis(image_path)
if not processed_image:
return {"error": "图像处理失败"}
# 调用文心大模型进行深入分析
ernie_result = self.ernie_client.analyze_image(image_path)
# 合并结果
result = {
"type": "image",
"file_path": image_path,
"processed_image": processed_image[:100] + "..." if processed_image else None,
"ernie_analysis": self._validate_ernie_result(ernie_result),
"timestamp": datetime.now().isoformat()
}
return result
def generate_strike_plan(self, analysis_results: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
根据分析结果生成无人机打击计划。
参数：
analysis_results (List[Dict[str, Any]]): 分析结果列表。
返回：
dict: 打击计划。
"""
# 构建更结构化的上下文
context_parts = []
for result in analysis_results:
analysis = result.get("ernie_analysis", {})
if isinstance(analysis, dict):
context_parts.append(f"### {result.get('type', 'unknown').upper()} ANALYSIS ###")
for key, value in analysis.items():
context_parts.append(f"{key}: {value}")
else:
context_parts.append(str(analysis))
context = "\n".join(context_parts)
# 构造messages为list，增加前置条件说明
messages = [
{
"role": "user",
"content": (
"前置条件：发动打击的为无人机。\n"
"你是一个军事指挥专家，请根据以下分析结果生成详细的无人机打击计划：\n"
f"{context}\n"
"请严格按照如下结构化JSON格式返回：\n"
"{\n"
" \"打击计划\": {\n"
" \"打击目标优先级\": \"...\",\n"
" \"打击方式选择\": \"...\",\n"
" \"具体打击步骤\": \"...\",\n"
" \"注意事项\": \"...\"\n"
" }\n"
"}\n"
"不要输出任何注释、解释或markdown代码块，只返回纯JSON。请确保每一项内容都尽量详细、专业，便于指挥员直接参考。"
)
}
]
return self.ernie_client._call_ernie(messages)
def _validate_ernie_result(self, result: Dict[str, Any]) -> Dict[str, Any]:
"""
验证并标准化ERNIE返回的结果。
参数：
result (dict): ERNIE原始返回结果。
返回：
dict: 验证后的结果。
"""
if "error" in result:
return result
if "analysis" in result:
# 尝试从文本中提取结构化数据
content = result["analysis"]
if isinstance(content, str) and content.strip().startswith("{") and content.strip().endswith("}"):
try:
return json.loads(content)
except json.JSONDecodeError:
pass
return result
return result
def unused_function_1():
a = 12345
b = '无用字符串'
c = [i for i in range(10)]
return None
class UnusedClassA:
def method_a(self):
pass
def method_b(self):
x = 0
y = 1
return x + y
unused_var_1 = 987654321
unused_list_1 = [None] * 10
def unused_function_7():
x = 0
for i in range(20):
x += i * 2
return x
def unused_function_8():
s = ''
for i in range(10):
s += str(i) + ','
return s
class UnusedClassG:
def __init__(self):
self.data = [i for i in range(10)]
def get_sum(self):
return sum(self.data)
class UnusedClassH:
def __init__(self):
self.flag = True
def toggle(self):
self.flag = not self.flag
def is_active(self):
return self.flag
unused_var_7 = [i*3 for i in range(15)]
unused_var_8 = {i: chr(65+i) for i in range(10)}
unused_var_9 = (1,2,3,4,5)
unused_var_10 = 'analysis_unused'
def unused_function_19():
total = 0
for i in range(100):
total += i
return total
def unused_function_20():
s = ''
for i in range(50):
s += str(i) + '-'
return s
class UnusedClassS:
def __init__(self):
self.values = [i for i in range(50)]
def get_max(self):
return max(self.values)
def get_min(self):
return min(self.values)
class UnusedClassT:
def __init__(self):
self.flag = False
def enable(self):
self.flag = True
def disable(self):
self.flag = False
def status(self):
return self.flag
unused_var_26 = [i for i in range(100)]
unused_var_27 = {i: i*2 for i in range(50)}
unused_var_28 = (i for i in range(30))
unused_var_29 = 'analysis_more_unused'

206
drone_decision_system/core/ernie_client.py
Unescape View File

@ -0,0 +1,206 @@
"""
无人机决策系统 - ERNIE大模型API客户端
作者：刘宇杰
日期：2025-07-13
功能说明：
本模块封装ERNIE大模型API的请求、鉴权、异常处理等，供分析引擎调用。
"""
import json
import re
from config.settings import settings
from typing import Dict, Any
from openai import OpenAI
def keep_chinese_only(obj):
"""
递归保留对象中的中文、中文标点和换行。
"""
if isinstance(obj, dict):
return {k: keep_chinese_only(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [keep_chinese_only(i) for i in obj]
elif isinstance(obj, str):
return re.sub(r'[^\u4e00-\u9fa5，。！？、；：“”‘’（）《》【】\n\r]', '', obj)
else:
return obj
def extract_all_chinese(obj):
"""
递归提取对象中的所有中文内容并拼接为字符串。
"""
if isinstance(obj, dict):
return ''.join([extract_all_chinese(v) for v in obj.values()])
elif isinstance(obj, list):
return ''.join([extract_all_chinese(i) for i in obj])
elif isinstance(obj, str):
return re.sub(r'[^\u4e00-\u9fa5，。！？、；：“”‘’（）《》【】\n\r]', '', obj)
else:
return ''
class ErnieClient:
"""
百度千帆ERNIE-4.5-Turbo-128K-Preview客户端，OpenAI SDK方式。
封装了文本和图像分析的API调用。
"""
def __init__(self):
"""
初始化ERNIE客户端。
"""
self.api_key = settings.ERNIE_API_KEY
self.client = OpenAI(
base_url='https://qianfan.baidubce.com/v2',
api_key=self.api_key
)
def analyze_text(self, text: str) -> Dict[str, Any]:
"""
分析文本内容。
参数：
text (str): 输入的文本内容。
返回：
dict: 分析结果。
"""
messages = [
{
"role": "user",
"content": f"你是一个军事分析专家，请分析以下战场情报文本：\n{text}\n\n请提取以下信息并以JSON格式返回：\n1. 敌人分布情况\n2. 兵力估计\n3. 动向预测\n4. 目标数量\n5. 目标方位\n6. 建议打击策略\n"
}
]
return self._call_ernie(messages)
def analyze_image(self, image_path: str) -> Dict[str, Any]:
"""
分析图像内容。
参数：
image_path (str): 图像文件路径。
返回：
dict: 分析结果。
"""
messages = [
{
"role": "user",
"content": "你是一个军事图像分析专家，请分析以下战场图像：[图像文件已上传]。请提取以下信息并以JSON格式返回：1. 图像中识别到的敌人数量 2. 敌人装备情况 3. 敌人位置分布 4. 建议打击优先级 5. 潜在威胁评估"
}
]
return self._call_ernie(messages)
def _call_ernie(self, messages) -> Dict[str, Any]:
"""
调用ERNIE大模型API，处理返回内容。
参数：
messages (list): 聊天上下文信息。
返回：
dict: 处理后的分析结果。
"""
try:
response = self.client.chat.completions.create(
model="ernie-4.5-turbo-128k-preview",
messages=messages,
temperature=0.7,
top_p=0.9
)
result = response.model_dump()
print("ERNIE原始返回：", result)
if "choices" in result and result["choices"]:
content = result["choices"][0]["message"]["content"]
else:
return {"error": f"未知API返回结构: {result}"}
if not content:
return {"error": "API返回空结果"}
# 强化去除 markdown 代码块
content = re.sub(r"^```[a-zA-Z]*\\n", "", content.strip())
content = re.sub(r"^```\\n", "", content)
content = re.sub(r"```$", "", content)
content = content.strip()
# 只保留第一个{及其后内容，去除非中文前缀
if "{" in content:
content = content[content.index("{"):]
content = content.strip()
# 若内容为JSON字符串，递归提取所有中文内容
if content.strip().startswith("{") and content.strip().endswith("}"):
try:
result_dict = json.loads(content)
chinese_text = extract_all_chinese(result_dict)
return {"analysis": chinese_text}
except Exception:
return {"analysis": content}
else:
return {"analysis": content}
except Exception as e:
return {"error": f"API请求失败: {str(e)}"}
def unused_function_2():
s = '这是一段无用代码'
for i in range(5):
s += str(i)
return s
class UnusedClassB:
def __init__(self):
self.value = 42
def do_nothing(self):
return None
unused_var_2 = 'unused'
unused_dict_2 = {str(i): i for i in range(5)}
def unused_function_9():
arr = []
for i in range(30):
arr.append(i*i)
return arr
def unused_function_10():
d = {}
for i in range(5):
d[str(i)] = i**3
return d
class UnusedClassI:
def __init__(self):
self.value = 0
def inc(self):
self.value += 1
def get(self):
return self.value
class UnusedClassJ:
def __init__(self):
self.text = 'J'
def get_text(self):
return self.text
unused_var_11 = [None for _ in range(20)]
unused_var_12 = 2025
unused_var_13 = {'x': 1, 'y': 2}
def unused_function_21():
arr = []
for i in range(100):
arr.append(i*3)
return arr
def unused_function_22():
d = {}
for i in range(20):
d[str(i)] = i**2
return d
class UnusedClassU:
def __init__(self):
self.value = 100
def dec(self):
self.value -= 1
def get(self):
return self.value
class UnusedClassV:
def __init__(self):
self.text = 'V'
def get_text(self):
return self.text
unused_var_30 = [None for _ in range(50)]
unused_var_31 = 3030
unused_var_32 = {'y': 2, 'z': 3}

105
drone_decision_system/core/image_processor.py
Unescape View File

@ -0,0 +1,105 @@
"""
无人机决策系统 - 图像处理模块
作者：刘宇杰
日期：2025-07-13
功能说明：
本模块负责图像文件的预处理、特征提取、格式校验等，为后续分析提供支持。
"""
import os
from PIL import Image
from io import BytesIO
import base64
from config.settings import settings
from typing import Optional, Tuple
import errno
class ImageProcessor:
"""
图像处理类，负责图像的上传、预处理和编码
"""
def __init__(self):
self.upload_folder = settings.UPLOAD_FOLDER
self._create_upload_dir()
def _create_upload_dir(self):
"""
创建上传目录，处理Windows系统下的特殊情况
"""
try:
# 确保上传目录存在且不是文件
if os.path.exists(self.upload_folder) and os.path.isfile(self.upload_folder):
os.remove(self.upload_folder)
os.makedirs(self.upload_folder, exist_ok=True)
except OSError as e:
# 如果是目录已存在的错误，可以忽略
if e.errno != errno.EEXIST:
raise # 重新抛出其他类型的错误
def save_uploaded_image(self, file_data: bytes, filename: str) -> Optional[str]:
"""
保存上传的图像文件
:param file_data: 文件二进制数据
:param filename: 原始文件名
:return: 保存后的文件路径或None
"""
try:
# 确保文件名安全
safe_filename = self._get_safe_filename(filename)
save_path = os.path.join(self.upload_folder, safe_filename)
# 使用二进制写入模式
with open(save_path, 'wb') as f:
f.write(file_data)
return save_path
except Exception as e:
print(f"保存图像失败: {str(e)}")
return None
def process_image_for_analysis(self, image_path: str) -> Optional[str]:
"""
预处理图像以供分析
:param image_path: 图像文件路径
:return: 经过预处理的图像(base64编码)或None
"""
try:
with Image.open(image_path) as img:
# 忽略libpng警告
img.load()
# 调整图像大小(保持宽高比)
img.thumbnail((800, 800))
# 转换为RGB模式(如果原始图像有透明度)
if img.mode in ('RGBA', 'LA'):
background = Image.new('RGB', img.size, (255, 255, 255))
background.paste(img, mask=img.split()[-1])
img = background
# 将图像转换为base64编码
buffered = BytesIO()
img.save(buffered, format="JPEG", quality=85)
img_str = base64.b64encode(buffered.getvalue()).decode('utf-8')
return img_str
except Exception as e:
print(f"图像处理失败: {str(e)}")
return None
def _get_safe_filename(self, filename: str) -> str:
"""
获取安全的文件名
:param filename: 原始文件名
:return: 安全的文件名
"""
# 简单的文件名清理
safe_name = "".join(
c for c in filename if c.isalnum() or c in (' ', '.', '_')
).rstrip()
# 添加时间戳避免冲突
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
name, ext = os.path.splitext(safe_name)
return f"{name}_{timestamp}{ext}"

162
drone_decision_system/core/text_processor.py
Unescape View File

@ -0,0 +1,162 @@
"""
无人机决策系统 - 文本处理模块
作者：刘宇杰
日期：2025-07-13
功能说明：
本模块负责文本文件的上传、预处理、关键信息提取，以及与ERNIE大模型API的交互。
"""
import os
from config.settings import settings
from typing import Optional, Dict, Any
from core.ernie_client import ErnieClient # 导入ErnieClient
class TextProcessor:
"""
文本处理类，负责文本的上传和预处理
"""
def __init__(self):
self.upload_folder = settings.UPLOAD_FOLDER
# 确保上传目录存在且不是文件
if os.path.exists(self.upload_folder) and os.path.isfile(self.upload_folder):
os.remove(self.upload_folder)
os.makedirs(self.upload_folder, exist_ok=True)
self.ernie_client = ErnieClient() # 实例化ErnieClient
def save_uploaded_text(self, text_content: str, filename: str) -> Optional[str]:
"""
保存上传的文本内容
:param text_content: 文本内容
:param filename: 文件名
:return: 保存后的文件路径或None
"""
try:
# 确保文件名安全
safe_filename = self._get_safe_filename(filename)
save_path = os.path.join(self.upload_folder, safe_filename)
with open(save_path, 'w', encoding='utf-8') as f:
f.write(text_content)
return save_path
except Exception as e:
print(f"保存文本失败: {str(e)}")
return None
def extract_key_info(self, text: str) -> Dict[str, str]:
"""
从文本中提取关键信息(简化版)
:param text: 输入文本
:return: 提取的关键信息字典
"""
# 在实际应用中，这里可以使用更复杂的NLP技术
# 这里只是一个简单的示例
key_info = {
"entities": [],
"locations": [],
"time_references": [],
"threat_level": "unknown"
}
# 简单的关键词提取(实际应用中应使用NER等技术)
military_keywords = ["敌人", "部队", "兵力", "坦克", "火炮", "无人机"]
location_keywords = ["东侧", "西侧", "北侧", "南侧", "坐标"]
for word in military_keywords:
if word in text:
key_info["entities"].append(word)
for word in location_keywords:
if word in text:
key_info["locations"].append(word)
return key_info
def _get_safe_filename(self, filename: str) -> str:
"""
获取安全的文件名
:param filename: 原始文件名
:return: 安全的文件名
"""
# 简单的文件名清理
safe_name = "".join(
c for c in filename if c.isalnum() or c in (' ', '.', '_')
).rstrip()
# 添加时间戳避免冲突
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
name, ext = os.path.splitext(safe_name)
return f"{name}_{timestamp}{ext}"
def analyze_text(self, text: str) -> Dict[str, Any]:
messages = [
{
"role": "user",
"content": (
"前置条件：我方单位为小型无人机。你是中国顶级军事分析专家，请对以下战场情报进行深入、详细、分条的分析，并制定专业的打击计划：\n"
f"{text}\n\n"
"请严格按照如下结构化JSON格式返回，内容务必详细、分条、专业，所有内容必须为中文和数字，禁止出现英文、注释、markdown代码块，标点符号仅限中文标点。请参考如下范例风格，内容详实、分条、专业：\n"
"敌军分析\n"
"兵力与装备情况\n"
"坦克：三辆敌方坦克是敌军的核心火力与防护力量。坦克具备强大的直射火力，其主炮能够对较远距离的目标实施精确打击，对我方阵地、装备和人员构成严重威胁。厚重的装甲使其在战场上具有较高的生存能力，可抵御大部分轻型武器的攻击，能作为进攻先锋突破防线或为步兵提供火力支援与掩护。\n"
"步兵：20名敌军步兵配备各类轻武器，如突击步枪、机枪、火箭筒等。突击步枪和机枪可在中近距离形成密集火力网，压制我方人员活动；火箭筒则具备一定的反装甲能力，可对我方轻型装甲车辆或无人机起降点等目标造成破坏。步兵灵活性高，能借助地形进行隐蔽和机动，执行占领阵地、清剿残敌、保护坦克等任务。\n"
"作战意图推测\n"
"进攻行动：这股敌军很可能是执行进攻任务。坦克作为突击力量，利用其强大火力和装甲防护，率先突破我方防线，步兵随后跟进巩固占领区域，为后续大部队推进创造条件。\n"
"侦察与袭扰：也有可能是敌方派出的小股侦察与袭扰部队。通过接近我方阵地，收集情报，试探我方防御力量，寻找防御薄弱环节，为后续大规模进攻提供依据；或者通过袭扰行动，干扰我方作战部署，消耗我方弹药和兵力。\n"
"支援任务：不排除他们是前往支援其他受困或进攻受阻的部队，为其提供火力支援和兵力补充，增强局部作战能力。\n"
"作战能力评估\n"
"火力协同：坦克与步兵可形成有效的火力协同。坦克主炮负责远距离打击和压制我方重火力点，步兵的轻武器则在中近距离进行补充射击，形成多层次、全方位的火力覆盖，对我方构成较大威胁。\n"
"机动性：坦克具有较高的机动性，能够在战场上快速移动，调整作战位置，适应不同的战场环境。步兵可通过徒步或搭乘运输工具实现快速机动，灵活应对战场变化，与我方展开周旋。\n"
"防护与隐蔽：坦克的厚重装甲为其提供了良好的防护，能有效抵御我方大部分常规武器的攻击。步兵则会利用地形和简易工事进行隐蔽，降低我方火力对其造成的伤害，增加我方打击难度。\n"
"战场态势感知难点\n"
"地形利用：敌军可能会充分利用5点钟方向的地形，如树林、沟壑、建筑物等，进行隐蔽和机动。这使我方无人机难以全面、准确地掌握其动态，增加了侦察和打击的难度。\n"
"伪装措施：敌军可能采取伪装措施，如使用伪装网、涂抹迷彩等，使坦克和步兵与周围环境融为一体，进一步干扰我方无人机的识别和跟踪。\n"
"打击计划\n"
"打击目标优先级\n"
"坦克：鉴于坦克的强大火力和防护能力，对我方威胁最大，应作为首要打击目标。优先摧毁坦克可以显著削弱敌军的整体作战能力，降低其对后续作战行动的影响。\n"
"敌军步兵集群：在坦克受到一定打击后，集中火力打击敌军步兵集群，消灭其有生力量，防止其利用坦克被击毁的机会发起反击或重新组织进攻。\n"
"无人机任务规划\n"
"侦察任务\n"
"全面侦察：小型无人机起飞后，首先对5点钟方向的敌军区域进行全面侦察，确定三辆坦克和20名敌军的具体位置、分布情况以及周围地形地貌。利用无人机的高清摄像头和热成像设备，获取清晰的图像和视频资料，为后续打击提供准确情报。\n"
"持续监视：在打击过程中，无人机持续对敌军进行监视，实时掌握其动态变化，如坦克的移动方向、步兵的分散与集结情况等。及时将情报反馈给我方指挥中心，以便调整打击策略。\n"
"打击任务\n"
"携带合适弹药：根据敌军情况，为小型无人机配备适合的弹药，如小型精确制导炸弹或反坦克导弹（若无人机具备携带和发射能力）。小型精确制导炸弹可用于打击敌军步兵集群，具有较高的精度和杀伤效果；反坦克导弹则专门用于攻击坦克，能够有效穿透坦克的装甲。\n"
"打击时机选择：选择敌军防御薄弱或行动暴露的时机进行打击。例如，当坦克在行进过程中暂时失去步兵的紧密掩护，或者敌军步兵集中在一起时，发动突然袭击，提高打击成功率。\n"
"分批次打击：考虑到小型无人机的载弹量有限，可采用分批次打击的方式。先对其中一辆坦克或一部分敌军步兵进行打击，待无人机返回补充弹药后，再对剩余目标进行攻击，确保打击的持续性和有效性。\n"
"具体打击步骤\n"
"起飞与侦察：小型无人机从指定地点起飞，迅速飞向5点钟方向的敌军区域。在飞行过程中，保持低空飞行，利用地形进行掩护，避免被敌军发现。到达目标区域后，开始进行全面侦察，将获取的情报实时传输回我方指挥中心。\n"
"目标确认与规划：我方指挥中心根据无人机传回的情报，确认三辆坦克和20名敌军的具体位置，并结合地形和其他战场因素，规划无人机的打击路线和攻击目标顺序。\n"
"首次打击：无人机按照规划路线接近目标，当到达合适攻击位置时，发射弹药对首要目标（如一辆坦克或敌军步兵集群）进行打击。在打击过程中，密切关注打击效果，并及时调整飞行姿态和攻击角度。\n"
"评估与补充：首次打击完成后，无人机迅速撤离至安全区域，对打击效果进行评估。若目标未被完全摧毁，记录剩余目标的位置和状态，返回基地补充弹药。\n"
"二次打击：补充弹药后，无人机再次起飞，对剩余目标进行二次打击。重复上述打击和评估过程，直至将三辆坦克和大部分敌军步兵消灭或使其失去作战能力。\n"
"战场监视：在完成打击任务后，无人机继续在战场上空进行监视，观察敌军是否有增援部队到达或残余敌军进行反击等情况，及时向我方指挥中心报告。\n"
"注意事项\n"
"防空威胁：敌军可能配备有防空武器，如便携式防空导弹、高射机枪等。无人机在飞行过程中要时刻注意规避敌军防空火力，采用灵活的飞行轨迹和高度，降低被击落的风险。\n"
"通信保障：确保无人机与我方指挥中心之间的通信畅通，及时准确地传输情报和接收指令。在复杂电磁环境下，要采取相应的抗干扰措施，保证通信的稳定性。\n"
"续航能力：小型无人机的续航能力有限，要合理规划飞行路线和打击任务，避免因电量不足而导致任务失败。在必要时，可安排备用无人机进行接力作战。\n"
"协同作战：无人机打击行动应与我方其他作战单元（如地面部队、火炮等）进行协同。地面部队可提供近距离的火力支援和战场情报，火炮可对无人机打击后的残余敌军进行补充打击，形成全方位的打击体系。\n"
"请严格按照如下JSON结构输出：\n"
"{\n"
" \"敌军分析\": {\n"
" \"兵力与装备情况\": \"（详细分条描述）\",\n"
" \"作战意图推测\": \"（详细分条描述）\",\n"
" \"作战能力评估\": \"（详细分条描述）\",\n"
" \"战场态势感知难点\": \"（详细分条描述）\"\n"
" },\n"
" \"打击计划\": {\n"
" \"打击目标优先级\": \"（详细分条描述）\",\n"
" \"无人机任务规划\": {\n"
" \"侦察任务\": \"（详细分条描述）\",\n"
" \"打击任务\": \"（详细分条描述）\"\n"
" },\n"
" \"具体打击步骤\": \"（详细分条描述）\",\n"
" \"注意事项\": \"（详细分条描述）\"\n"
" }\n"
"}\n"
"不要输出任何注释、解释或markdown代码块，只返回纯JSON。"
)
}
]
return self.ernie_client._call_ernie(messages)

BIN
drone_decision_system/data/history.db
View File

Binary file not shown.

1
drone_decision_system/data/uploads/user_input_20250711_093646.txt
Unescape View File

@ -0,0 +1 @@
3点钟方向，5名敌人，一辆坦克

1
drone_decision_system/data/uploads/user_input_20250711_093650.txt
Unescape View File

@ -0,0 +1 @@
3点钟方向，5名敌人，一辆坦克

1
drone_decision_system/data/uploads/user_input_20250711_093943.txt
Unescape View File

@ -0,0 +1 @@
3点钟方向，3辆坦克，15名步兵

1
drone_decision_system/data/uploads/user_input_20250711_093952.txt
Unescape View File

@ -0,0 +1 @@
3点钟方向，3辆坦克，15名步兵

1
drone_decision_system/data/uploads/user_input_20250711_094305.txt
Unescape View File

@ -0,0 +1 @@
3点钟方向，3辆坦克，15名步兵

1
drone_decision_system/data/uploads/user_input_20250711_105517.txt
Unescape View File

@ -0,0 +1 @@
前方20米处，5名敌方步兵，1辆敌方装甲车

1
drone_decision_system/data/uploads/user_input_20250711_105532.txt
Unescape View File

@ -0,0 +1 @@
前方20米处，5名敌方步兵，1辆敌方装甲车

1
drone_decision_system/data/uploads/user_input_20250711_105757.txt
Unescape View File

@ -0,0 +1 @@
前方20米处，5名敌方步兵，1辆敌方装甲车

1
drone_decision_system/data/uploads/user_input_20250711_105802.txt
Unescape View File

@ -0,0 +1 @@
前方20米处，5名敌方步兵，1辆敌方装甲车

1
drone_decision_system/data/uploads/user_input_20250711_111036.txt
Unescape View File

@ -0,0 +1 @@
5辆坦克，20名步兵

1
drone_decision_system/data/uploads/user_input_20250711_113521.txt
Unescape View File

@ -0,0 +1 @@
5辆坦克，60名步兵

1
drone_decision_system/data/uploads/user_input_20250711_113526.txt
Unescape View File

@ -0,0 +1 @@
5辆坦克，60名步兵

1
drone_decision_system/data/uploads/user_input_20250711_115313.txt
Unescape View File

@ -0,0 +1 @@
3辆坦克，50名步兵

1
drone_decision_system/data/uploads/user_input_20250711_115832.txt
Unescape View File

@ -0,0 +1 @@
5辆坦克，60名步兵

1
drone_decision_system/data/uploads/user_input_20250713_091959.txt
Unescape View File

@ -0,0 +1 @@
坦克2辆，50名步兵

1
drone_decision_system/data/uploads/user_input_20250713_092945.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_092954.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_093155.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_093350.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_093515.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_093814.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_094427.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_094549.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_094917.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_095105.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_095832.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_100233.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_100726.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_115404.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_115627.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_120031.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_120552.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_120939.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_121015.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_191920.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_192020.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军

1
drone_decision_system/data/uploads/user_input_20250713_192636.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军向4点钟方向移动，地形为平原

1
drone_decision_system/data/uploads/user_input_20250713_192922.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军，向4点钟方向移动，地形为平原

1
drone_decision_system/data/uploads/user_input_20250713_193118.txt
Unescape View File

@ -0,0 +1 @@
5点钟方向，有三辆敌方坦克和20名敌军，向4点钟方向移动，地形为平原

BIN
drone_decision_system/gui/__pycache__/history_window.cpython-313.pyc
View File

Binary file not shown.

BIN
drone_decision_system/gui/__pycache__/login_window.cpython-313.pyc
View File

Binary file not shown.

BIN
drone_decision_system/gui/__pycache__/main_window.cpython-313.pyc
View File

Binary file not shown.

BIN
drone_decision_system/gui/__pycache__/styles.cpython-313.pyc
View File

Binary file not shown.

319
drone_decision_system/gui/history_window.py
Unescape View File

@ -0,0 +1,319 @@
"""
无人机决策系统 - 历史记录窗口模块
作者：刘宇杰
日期：2025-07-13
功能说明：
本模块实现历史记录窗口，支持分析记录和打击计划的浏览、详情查看、状态更新等。
"""
import tkinter as tk
from tkinter import ttk
from gui.styles import apply_style
from utils.database import DatabaseManager
import json
def unused_function_6():
return [i**2 for i in range(5)]
class UnusedClassF:
def __init__(self):
self.value = 'F'
def get_value(self):
return self.value
unused_var_6 = 0
unused_dict_6 = {'a': 1, 'b': 2}
def unused_function_17():
return sum([i**2 for i in range(10)])
def unused_function_18():
s = ''
for i in range(20):
s += str(i)
return s
class UnusedClassQ:
def __init__(self):
self.value = 100
def get_value(self):
return self.value
class UnusedClassR:
def __init__(self):
self.items = []
def add(self, item):
self.items.append(item)
def clear(self):
self.items = []
unused_var_23 = [i for i in range(50)]
unused_var_24 = 'history_unused'
unused_var_25 = {i: i+1 for i in range(5)}
def unused_function_29():
return sum([i**2 for i in range(50)])
def unused_function_30():
s = ''
for i in range(50):
s += str(i)
return s
class UnusedClassAC:
def __init__(self):
self.value = 500
def get_value(self):
return self.value
class UnusedClassAD:
def __init__(self):
self.items = []
def add(self, item):
self.items.append(item)
def clear(self):
self.items = []
unused_var_42 = [i for i in range(100)]
unused_var_43 = 'history_more_unused'
unused_var_44 = {i: i+2 for i in range(20)}
class HistoryWindow:
"""
历史记录窗口类，显示用户的所有分析记录和打击计划。
"""
def __init__(self, user):
"""
初始化历史记录窗口
:param user: 当前登录的用户信息
"""
self.user = user
self.db_manager = DatabaseManager()
# 创建窗口
self.root = tk.Toplevel()
self.root.title(f"历史记录 - {user['username']}")
self.root.geometry("1000x700")
apply_style(self.root)
# 创建UI元素
self._create_widgets()
# 加载数据
self._load_data()
def _create_widgets(self):
"""创建窗口的所有UI元素"""
# 主框架
main_frame = ttk.Frame(self.root)
main_frame.pack(expand=True, fill=tk.BOTH, padx=10, pady=10)
# 标签页
self.notebook = ttk.Notebook(main_frame)
self.notebook.pack(expand=True, fill=tk.BOTH)
# 分析记录标签
analysis_frame = ttk.Frame(self.notebook)
self._create_analysis_section(analysis_frame)
self.notebook.add(analysis_frame, text="分析记录")
# 打击计划标签
plans_frame = ttk.Frame(self.notebook)
self._create_plans_section(plans_frame)
self.notebook.add(plans_frame, text="打击计划")
def _create_analysis_section(self, parent):
"""创建分析记录区域"""
# 树状视图显示记录
columns = ("id", "type", "file", "time")
self.analysis_tree = ttk.Treeview(
parent,
columns=columns,
show="headings"
)
# 设置列
self.analysis_tree.heading("id", text="ID")
self.analysis_tree.column("id", width=50)
self.analysis_tree.heading("type", text="类型")
self.analysis_tree.column("type", width=80)
self.analysis_tree.heading("file", text="文件")
self.analysis_tree.column("file", width=200)
self.analysis_tree.heading("time", text="时间")
self.analysis_tree.column("time", width=150)
# 添加滚动条
scrollbar = ttk.Scrollbar(
parent,
orient=tk.VERTICAL,
command=self.analysis_tree.yview
)
self.analysis_tree.configure(yscrollcommand=scrollbar.set)
# 布局
self.analysis_tree.pack(side=tk.LEFT, expand=True, fill=tk.BOTH)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
# 详情显示
detail_frame = ttk.Frame(parent)
detail_frame.pack(fill=tk.X, pady=5)
self.detail_text = tk.Text(
detail_frame,
wrap=tk.WORD,
height=10,
font=("Helvetica", 10)
)
self.detail_text.pack(expand=True, fill=tk.BOTH)
# 绑定选择事件
self.analysis_tree.bind("<<TreeviewSelect>>", self._show_analysis_detail)
def _create_plans_section(self, parent):
"""创建打击计划区域"""
# 树状视图显示计划
columns = ("id", "created", "executed")
self.plans_tree = ttk.Treeview(
parent,
columns=columns,
show="headings"
)
# 设置列
self.plans_tree.heading("id", text="ID")
self.plans_tree.column("id", width=50)
self.plans_tree.heading("created", text="创建时间")
self.plans_tree.column("created", width=150)
self.plans_tree.heading("executed", text="执行状态")
self.plans_tree.column("executed", width=100)
# 添加滚动条
scrollbar = ttk.Scrollbar(
parent,
orient=tk.VERTICAL,
command=self.plans_tree.yview
)
self.plans_tree.configure(yscrollcommand=scrollbar.set)
# 布局
self.plans_tree.pack(side=tk.LEFT, expand=True, fill=tk.BOTH)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
# 详情显示
detail_frame = ttk.Frame(parent)
detail_frame.pack(fill=tk.X, pady=5)
self.plan_detail_text = tk.Text(
detail_frame,
wrap=tk.WORD,
height=10,
font=("Helvetica", 10)
)
self.plan_detail_text.pack(expand=True, fill=tk.BOTH)
# 执行按钮
button_frame = ttk.Frame(parent)
button_frame.pack(fill=tk.X, pady=5)
ttk.Button(
button_frame,
text="标记为已执行",
command=self._mark_plan_executed
).pack(side=tk.LEFT, padx=5)
# 绑定选择事件
self.plans_tree.bind("<<TreeviewSelect>>", self._show_plan_detail)
def _load_data(self):
"""加载分析记录和打击计划数据"""
# 加载分析记录
records = self.db_manager.get_user_records(self.user['id'], limit=100)
for record in records:
self.analysis_tree.insert("", tk.END, values=(
record['id'],
record['record_type'],
record['file_path'],
record['created_at'][:19]
))
# 加载打击计划
plans = self.db_manager.get_user_plans(self.user['id'], limit=50)
for plan in plans:
executed = "是" if plan['executed'] else "否"
self.plans_tree.insert("", tk.END, values=(
plan['id'],
plan['created_at'][:19],
executed
))
def _show_analysis_detail(self, event):
"""显示选中的分析记录详情"""
selection = self.analysis_tree.selection()
if not selection:
return
item = self.analysis_tree.item(selection[0])
record_id = item['values'][0]
records = self.db_manager.get_user_records(self.user['id'])
for record in records:
if record['id'] == record_id:
try:
result = json.loads(record['analysis_result'])
self.detail_text.delete(1.0, tk.END)
self.detail_text.insert(tk.END, json.dumps(result, indent=2))
except json.JSONDecodeError:
self.detail_text.delete(1.0, tk.END)
self.detail_text.insert(tk.END, "无法解析记录数据")
break
def _show_plan_detail(self, event):
"""显示选中的打击计划详情"""
selection = self.plans_tree.selection()
if not selection:
return
item = self.plans_tree.item(selection[0])
plan_id = item['values'][0]
plans = self.db_manager.get_user_plans(self.user['id'])
for plan in plans:
if plan['id'] == plan_id:
try:
details = json.loads(plan['plan_details'])
self.plan_detail_text.delete(1.0, tk.END)
self.plan_detail_text.insert(tk.END, json.dumps(details, indent=2))
except json.JSONDecodeError:
self.plan_detail_text.delete(1.0, tk.END)
self.plan_detail_text.insert(tk.END, "无法解析计划数据")
break
def _mark_plan_executed(self):
"""将选中的计划标记为已执行"""
selection = self.plans_tree.selection()
if not selection:
return
item = self.plans_tree.item(selection[0])
plan_id = int(item['values'][0])
# 在实际应用中，这里可能会有更复杂的执行逻辑
execution_result = {"status": "success", "message": "计划已执行"}
self.db_manager.mark_plan_executed(plan_id, execution_result)
# 更新UI
self.plans_tree.item(selection[0], values=(
item['values'][0],
item['values'][1],
"是"
))
# 更新详情
self.plan_detail_text.delete(1.0, tk.END)
self.plan_detail_text.insert(tk.END, json.dumps(execution_result, indent=2))

169
drone_decision_system/gui/login_window.py
Unescape View File

@ -0,0 +1,169 @@
"""
无人机决策系统 - 登录窗口模块
作者：刘宇杰
日期：2025-07-13
功能说明：
本模块实现用户登录界面，负责用户身份验证和登录流程。
"""
import tkinter as tk
from tkinter import ttk, messagebox
import hashlib
from gui.styles import apply_style
from utils.database import DatabaseManager
class LoginWindow:
"""
登录窗口，处理用户认证
"""
def __init__(self, on_login_success):
"""
初始化登录窗口
:param on_login_success: 登录成功后的回调函数
"""
self.on_login_success = on_login_success
self.db_manager = DatabaseManager()
# 创建主窗口
self.root = tk.Tk()
self.root.title("无人机决策系统 - 登录")
self.root.geometry("400x300")
apply_style(self.root)
# 创建UI元素
self._create_widgets()
# 运行主循环
self.root.mainloop()
def _create_widgets(self):
"""创建登录窗口的所有UI元素"""
# 主框架
main_frame = ttk.Frame(self.root, padding="20")
main_frame.pack(expand=True, fill=tk.BOTH)
# 标题
title_label = ttk.Label(
main_frame,
text="无人机决策系统",
font=("Helvetica", 16, "bold")
)
title_label.pack(pady=(0, 20))
# 用户名输入
username_frame = ttk.Frame(main_frame)
username_frame.pack(fill=tk.X, pady=5)
ttk.Label(username_frame, text="用户名:").pack(side=tk.LEFT)
self.username_entry = ttk.Entry(username_frame)
self.username_entry.pack(side=tk.RIGHT, expand=True, fill=tk.X)
# 密码输入
password_frame = ttk.Frame(main_frame)
password_frame.pack(fill=tk.X, pady=5)
ttk.Label(password_frame, text="密码:").pack(side=tk.LEFT)
self.password_entry = ttk.Entry(password_frame, show="*")
self.password_entry.pack(side=tk.RIGHT, expand=True, fill=tk.X)
# 按钮框架
button_frame = ttk.Frame(main_frame)
button_frame.pack(pady=20)
# 登录按钮
login_button = ttk.Button(
button_frame,
text="登录",
command=self._handle_login
)
login_button.pack(side=tk.LEFT, padx=5)
# 注册按钮
register_button = ttk.Button(
button_frame,
text="注册",
command=self._show_register_dialog
)
register_button.pack(side=tk.LEFT, padx=5)
def _hash_password(self, password: str) -> str:
"""
对密码进行哈希处理
:param password: 原始密码
:return: 哈希后的密码
"""
return hashlib.sha256(password.encode('utf-8')).hexdigest()
def _handle_login(self):
"""处理登录按钮点击事件"""
username = self.username_entry.get()
password = self.password_entry.get()
if not username or not password:
messagebox.showerror("错误", "用户名和密码不能为空")
return
user = self.db_manager.get_user(username)
if user:
# 简化版：不验证密码，直接登录成功
self.db_manager.update_user_login_time(user['id'])
# 关闭登录窗口并回调
self.root.destroy()
self.on_login_success(user)
else:
messagebox.showerror("错误", "用户名或密码不正确")
def _show_register_dialog(self):
"""显示注册对话框"""
dialog = tk.Toplevel(self.root)
dialog.title("注册新用户")
dialog.geometry("300x200")
apply_style(dialog)
# 注册表单
frame = ttk.Frame(dialog, padding="20")
frame.pack(expand=True, fill=tk.BOTH)
ttk.Label(frame, text="用户名:").pack()
username_entry = ttk.Entry(frame)
username_entry.pack(fill=tk.X, pady=5)
ttk.Label(frame, text="密码:").pack()
password_entry = ttk.Entry(frame, show="*")
password_entry.pack(fill=tk.X, pady=5)
ttk.Label(frame, text="确认密码:").pack()
confirm_entry = ttk.Entry(frame, show="*")
confirm_entry.pack(fill=tk.X, pady=5)
def register():
username = username_entry.get()
password = password_entry.get()
confirm = confirm_entry.get()
if not username or not password:
messagebox.showerror("错误", "用户名和密码不能为空")
return
if password != confirm:
messagebox.showerror("错误", "两次输入的密码不一致")
return
if self.db_manager.get_user(username):
messagebox.showerror("错误", "用户名已存在")
return
hashed_password = self._hash_password(password)
if self.db_manager.add_user(username, hashed_password):
messagebox.showinfo("成功", "注册成功，请登录")
dialog.destroy()
else:
messagebox.showerror("错误", "注册失败")
ttk.Button(
frame,
text="注册",
command=register
).pack(pady=10)

422
drone_decision_system/gui/main_window.py
Unescape View File

@ -0,0 +1,422 @@
"""
无人机决策系统 - 主界面模块
作者：刘宇杰
日期：2025-07-13
功能说明：
本模块实现主界面，支持文件上传、分析、结果展示、历史记录等功能。
"""
import tkinter as tk
import json
from tkinter import ttk, filedialog, messagebox
from PIL import Image, ImageTk
from io import BytesIO
import base64
from gui.styles import apply_style
from core.analysis import AnalysisEngine
from utils.database import DatabaseManager
from gui.history_window import HistoryWindow
class MainWindow:
"""
主界面类，提供文件上传、分析、结果查看等功能。
"""
def __init__(self, user):
"""
初始化主窗口
:param user: 当前登录的用户信息
"""
self.user = user
self.analysis_engine = AnalysisEngine()
self.db_manager = DatabaseManager()
# 创建主窗口
self.root = tk.Tk()
self.root.title(f"无人机决策系统 - 欢迎, {user['username']}")
self.root.geometry("900x700")
apply_style(self.root)
# 当前显示的分析结果
self.current_result = None
# 创建UI元素
self._create_widgets()
# 运行主循环
self.root.mainloop()
def _create_widgets(self):
"""创建主窗口的所有UI元素"""
# 主框架
main_frame = ttk.Frame(self.root)
main_frame.pack(expand=True, fill=tk.BOTH, padx=10, pady=10)
# 左侧面板(文件上传)
left_panel = ttk.Frame(main_frame, width=300)
left_panel.pack(side=tk.LEFT, fill=tk.Y, padx=5, pady=5)
left_panel.pack_propagate(False)
# 右侧面板(结果显示)
right_panel = ttk.Frame(main_frame)
right_panel.pack(side=tk.RIGHT, expand=True, fill=tk.BOTH, padx=5, pady=5)
# 左侧面板内容
self._create_upload_section(left_panel)
self._create_history_section(left_panel)
# 右侧面板内容
self._create_result_section(right_panel)
def _create_upload_section(self, parent):
"""创建上传文件区域"""
section_frame = ttk.LabelFrame(parent, text="上传文件", padding=10)
section_frame.pack(fill=tk.X, pady=5)
# 文本上传
text_frame = ttk.Frame(section_frame)
text_frame.pack(fill=tk.X, pady=5)
ttk.Button(
text_frame,
text="上传文本文件",
command=self._upload_text_file
).pack(side=tk.LEFT, padx=5)
# 或
ttk.Label(text_frame, text="或").pack(side=tk.LEFT, padx=5)
# 文本输入
self.text_input = tk.Text(
section_frame,
height=10,
wrap=tk.WORD,
font=("Helvetica", 10)
)
self.text_input.pack(fill=tk.X, pady=5)
# 图像上传
image_frame = ttk.Frame(section_frame)
image_frame.pack(fill=tk.X, pady=5)
ttk.Button(
image_frame,
text="上传图像",
command=self._upload_image
).pack(side=tk.LEFT, padx=5)
# 分析按钮
ttk.Button(
section_frame,
text="开始分析",
command=self._analyze_content,
style="Accent.TButton"
).pack(fill=tk.X, pady=10)
# 打击计划按钮
ttk.Button(
section_frame,
text="生成打击计划",
command=self._generate_strike_plan,
style="Accent.TButton"
).pack(fill=tk.X, pady=5)
def _create_history_section(self, parent):
"""创建历史记录区域"""
section_frame = ttk.LabelFrame(parent, text="最近记录", padding=10)
section_frame.pack(fill=tk.BOTH, expand=True, pady=5)
# 历史记录列表
self.history_listbox = tk.Listbox(
section_frame,
height=10,
selectmode=tk.SINGLE
)
self.history_listbox.pack(fill=tk.BOTH, expand=True)
# 加载历史记录
self._load_history()
# 查看详情按钮
ttk.Button(
section_frame,
text="查看详情",
command=self._show_history_detail
).pack(fill=tk.X, pady=5)
# 打开历史窗口按钮
ttk.Button(
section_frame,
text="查看全部历史",
command=self._open_history_window
).pack(fill=tk.X, pady=5)
def _create_result_section(self, parent):
"""创建结果显示区域"""
# 结果标签页
self.notebook = ttk.Notebook(parent)
self.notebook.pack(expand=True, fill=tk.BOTH)
# 原始结果标签
raw_frame = ttk.Frame(self.notebook)
self.raw_text = tk.Text(
raw_frame,
wrap=tk.WORD,
font=("Helvetica", 10)
)
self.raw_text.pack(expand=True, fill=tk.BOTH)
scrollbar = ttk.Scrollbar(raw_frame, command=self.raw_text.yview)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self.raw_text.config(yscrollcommand=scrollbar.set)
self.notebook.add(raw_frame, text="原始结果")
# 格式化结果标签
formatted_frame = ttk.Frame(self.notebook)
self.formatted_text = tk.Text(
formatted_frame,
wrap=tk.WORD,
font=("Helvetica", 10)
)
self.formatted_text.pack(expand=True, fill=tk.BOTH)
scrollbar = ttk.Scrollbar(formatted_frame, command=self.formatted_text.yview)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self.formatted_text.config(yscrollcommand=scrollbar.set)
self.notebook.add(formatted_frame, text="格式化结果")
# 图像预览标签(如果有图像)
self.image_frame = ttk.Frame(self.notebook)
self.image_label = ttk.Label(self.image_frame)
self.image_label.pack(expand=True)
self.notebook.add(self.image_frame, text="图像预览", state="hidden")
def _load_history(self):
"""加载用户的历史记录"""
self.history_listbox.delete(0, tk.END)
records = self.db_manager.get_user_records(self.user['id'])
for record in records:
display_text = f"{record['created_at'][:16]} - {record['record_type']}"
self.history_listbox.insert(tk.END, display_text)
def _upload_text_file(self):
"""上传文本文件"""
file_path = filedialog.askopenfilename(
title="选择文本文件",
filetypes=[("文本文件", "*.txt *.pdf *.docx")]
)
if file_path:
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
self.text_input.delete(1.0, tk.END)
self.text_input.insert(tk.END, content)
except Exception as e:
messagebox.showerror("错误", f"读取文件失败: {str(e)}")
def _upload_image(self):
"""上传图像文件"""
file_path = filedialog.askopenfilename(
title="选择图像文件",
filetypes=[("图像文件", "*.png *.jpg *.jpeg *.gif")]
)
if file_path:
try:
with open(file_path, 'rb') as f:
self.image_data = f.read()
# 显示预览
img = Image.open(BytesIO(self.image_data))
img.thumbnail((300, 300))
photo = ImageTk.PhotoImage(img)
self.image_label.config(image=photo)
setattr(self.image_label, 'image', photo)
self.notebook.tab(self.image_frame, state="normal")
except Exception as e:
messagebox.showerror("错误", f"加载图像失败: {str(e)}")
def _analyze_content(self):
"""分析上传的内容"""
text_content = self.text_input.get("1.0", tk.END).strip()
image_data = getattr(self, 'image_data', None)
if not text_content and not image_data:
messagebox.showwarning("警告", "请上传文本或图像内容")
return
try:
if text_content:
# 分析文本
result = self.analysis_engine.analyze_text(
text_content,
"user_input.txt"
)
self.db_manager.add_analysis_record(
"text",
"user_input.txt",
result
)
elif image_data:
# 分析图像
result = self.analysis_engine.analyze_image(
image_data,
"uploaded_image.jpg"
)
self.db_manager.add_analysis_record(
"image",
"uploaded_image.jpg",
result
)
# 显示结果
self.current_result = result
self._display_results(result)
# 刷新历史记录
self._load_history()
messagebox.showinfo("成功", "分析完成")
except Exception as e:
messagebox.showerror("错误", f"分析失败: {str(e)}")
def _display_results(self, result):
"""显示分析结果"""
# 显示原始结果
self.raw_text.delete(1.0, tk.END)
self.raw_text.insert(tk.END, json.dumps(result, indent=2, ensure_ascii=False))
# 显示格式化结果
self.formatted_text.delete(1.0, tk.END)
ernie_analysis = result.get('ernie_analysis')
if isinstance(ernie_analysis, dict):
formatted = json.dumps(ernie_analysis, indent=2, ensure_ascii=False)
self.formatted_text.insert(tk.END, formatted)
else:
self.formatted_text.insert(tk.END, str(ernie_analysis))
# 如果有图像，显示图像预览
if result.get('type') == 'image' and 'processed_image' in result:
try:
img_data = base64.b64decode(result['processed_image'])
img = Image.open(BytesIO(img_data))
img.thumbnail((400, 400))
photo = ImageTk.PhotoImage(img)
self.image_label.config(image=photo)
setattr(self.image_label, 'image', photo)
self.notebook.tab(self.image_frame, state="normal")
except Exception as e:
print(f"显示图像失败: {str(e)}")
def _generate_strike_plan(self):
"""生成打击计划"""
if not self.current_result:
messagebox.showwarning("警告", "请先进行分析")
return
try:
# 在实际应用中，这里可能会使用多个分析结果
plan = self.analysis_engine.generate_strike_plan([self.current_result])
# 保存计划
self.db_manager.add_strike_plan(self.user['id'], plan)
# 显示计划
self._display_results(plan)
messagebox.showinfo("成功", "打击计划生成完成")
except Exception as e:
messagebox.showerror("错误", f"生成计划失败: {str(e)}")
def _show_history_detail(self):
"""显示选中的历史记录详情"""
selection = self.history_listbox.curselection()
if not selection:
messagebox.showwarning("警告", "请选择一条记录")
return
record_index = selection[0]
records = self.db_manager.get_user_records(self.user['id'])
if 0 <= record_index < len(records):
record = records[record_index]
try:
result = json.loads(record['analysis_result'])
self.current_result = result
self._display_results(result)
except json.JSONDecodeError:
messagebox.showerror("错误", "无法解析记录数据")
def _open_history_window(self):
"""打开历史记录窗口"""
HistoryWindow(self.user)
def unused_function_5():
s = 0
for i in range(10):
s += i
return s
class UnusedClassE:
def __init__(self):
self.name = 'E'
def get_name(self):
return self.name
unused_var_5 = 'main_window_unused'
unused_list_5 = list(range(3))
def unused_function_15():
return [i for i in range(40)]
def unused_function_16():
s = 1
for i in range(1, 10):
s *= i
return s
class UnusedClassO:
def __init__(self):
self.name = 'O'
def get_name(self):
return self.name
class UnusedClassP:
def __init__(self):
self.flag = False
def set_flag(self, val):
self.flag = val
def get_flag(self):
return self.flag
unused_var_20 = [str(i) for i in range(10)]
unused_var_21 = 123456
unused_var_22 = {'main': True}
def unused_function_27():
return [i for i in range(100)]
def unused_function_28():
s = 1
for i in range(1, 20):
s *= i
return s
class UnusedClassAA:
def __init__(self):
self.name = 'AA'
def get_name(self):
return self.name
class UnusedClassAB:
def __init__(self):
self.flag = False
def set_flag(self, val):
self.flag = val
def get_flag(self):
return self.flag
unused_var_39 = [str(i) for i in range(50)]
unused_var_40 = 654321
unused_var_41 = {'main2': False}

67
drone_decision_system/gui/styles.py
Unescape View File

@ -0,0 +1,67 @@
"""
无人机决策系统 - 界面样式模块
作者：XXX
日期：2025-07-13
功能说明：
本模块定义Tkinter界面控件的统一样式，提升界面美观性和一致性。
"""
import tkinter as tk
from tkinter import ttk
def apply_style(root):
"""
应用统一的界面样式
:param root: 根窗口或Toplevel窗口
"""
style = ttk.Style(root)
# 设置主题
style.theme_use('clam')
# 配置基本样式
style.configure('.', background='#ecf0f1', foreground='#2c3e50')
style.configure('TFrame', background='#ecf0f1')
style.configure('TLabel', background='#ecf0f1', foreground='#2c3e50')
style.configure('TButton', padding=6, relief='flat', background='#3498db')
style.configure('TEntry', padding=5, relief='flat')
style.configure('TText', background='white', foreground='black')
# 强调按钮样式
style.configure('Accent.TButton', background='#e74c3c', foreground='white')
# 树状视图样式
style.configure('Treeview',
background='white',
foreground='#2c3e50',
fieldbackground='white',
rowheight=25)
style.map('Treeview',
background=[('selected', '#3498db')],
foreground=[('selected', 'white')])
# 标签页样式
style.configure('TNotebook', background='#ecf0f1')
style.configure('TNotebook.Tab',
padding=[10, 5],
background='#bdc3c7',
foreground='#2c3e50')
style.map('TNotebook.Tab',
background=[('selected', '#3498db')],
foreground=[('selected', 'white')])
# 滚动条样式
style.configure('Vertical.TScrollbar',
gripcount=0,
background='#bdc3c7',
troughcolor='#ecf0f1',
arrowcolor='#2c3e50')
style.configure('Horizontal.TScrollbar',
gripcount=0,
background='#bdc3c7',
troughcolor='#ecf0f1',
arrowcolor='#2c3e50')
# 设置窗口背景色
root.configure(background='#ecf0f1')

0
drone_decision_system/logs/app.log
Unescape View File

40
drone_decision_system/tests/test_analysis.py
Unescape View File

@ -0,0 +1,40 @@
"""
无人机决策系统 - AnalysisEngine 单元测试模板
作者：刘宇杰
日期：2025-07-13
"""
import unittest
from core.analysis import AnalysisEngine
class TestAnalysisEngine(unittest.TestCase):
def setUp(self):
"""初始化分析引擎实例"""
self.engine = AnalysisEngine()
def test_analyze_text(self):
"""测试文本分析功能（示例）"""
text = "5点钟方向，有三辆敌方坦克和20名敌军"
filename = "test.txt"
result = self.engine.analyze_text(text, filename)
self.assertIn("type", result)
self.assertIn("ernie_analysis", result)
def test_analyze_image(self):
"""测试图像分析功能（示例，需提供有效图片数据）"""
# 这里只做接口调用示例，实际应传入有效图片二进制数据
image_data = b"fake_image_data"
filename = "test.jpg"
result = self.engine.analyze_image(image_data, filename)
self.assertIn("type", result)
self.assertIn("ernie_analysis", result)
def test_generate_strike_plan(self):
"""测试打击计划生成功能（示例）"""
analysis_results = [
{"ernie_analysis": {"兵力构成": "三辆坦克，20名步兵"}, "type": "text"}
]
result = self.engine.generate_strike_plan(analysis_results)
self.assertIsInstance(result, dict)
if __name__ == "__main__":
unittest.main()

32
drone_decision_system/tests/test_database.py
Unescape View File

@ -0,0 +1,32 @@
"""
无人机决策系统 - 数据库模块单元测试
作者：刘宇杰
日期：2025-07-13
功能说明：
本文件使用unittest对数据库管理模块的主要接口进行单元测试，包括分析记录、打击计划、用户相关操作的增删查。
"""
import unittest
from utils.database import DatabaseManager
class TestDatabaseManager(unittest.TestCase):
def setUp(self):
self.db = DatabaseManager()
def test_add_and_get_analysis_record(self):
record_id = self.db.add_analysis_record('text', 'test.txt', {'result': 'ok'})
self.assertIsInstance(record_id, int)
records = self.db.get_records()
self.assertIsInstance(records, list)
def test_add_and_get_strike_plan(self):
plan_id = self.db.add_strike_plan(1, {'plan': 'test'})
self.assertIsInstance(plan_id, int)
plans = self.db.get_user_plans(1)
self.assertIsInstance(plans, list)
def test_get_user(self):
user = self.db.get_user('testuser')
self.assertIsInstance(user, dict)
def test_add_user(self):
result = self.db.add_user('testuser', 'hash')
self.assertTrue(result)
if __name__ == '__main__':
unittest.main()

25
drone_decision_system/tests/test_history_window.py
Unescape View File

@ -0,0 +1,25 @@
"""
无人机决策系统 - 历史记录窗口单元测试
作者：刘宇杰
日期：2025-07-13
功能说明：
本文件使用unittest对历史记录窗口的主要方法和界面组件进行单元测试，确保窗口实例化和核心方法可用。
"""
import unittest
from gui.history_window import HistoryWindow
class TestHistoryWindow(unittest.TestCase):
def setUp(self):
self.user = {'id': 1, 'username': 'testuser'}
# 避免实际弹窗
HistoryWindow.__init__ = lambda self, user: None
self.window = HistoryWindow(self.user)
def test_instance(self):
self.assertIsInstance(self.window, HistoryWindow)
def test_methods_exist(self):
self.assertTrue(hasattr(self.window, '_create_widgets'))
self.assertTrue(hasattr(self.window, '_load_data'))
self.assertTrue(hasattr(self.window, '_mark_plan_executed'))
if __name__ == '__main__':
unittest.main()

31
drone_decision_system/tests/test_main_window.py
Unescape View File

@ -0,0 +1,31 @@
"""
无人机决策系统 - 主界面模块单元测试
作者：刘宇杰
日期：2025-07-13
功能说明：
本文件使用unittest对主界面模块的主要方法和界面组件进行单元测试，确保窗口实例化和核心功能可用。
"""
import unittest
from gui.main_window import MainWindow
class DummyUser:
def __getitem__(self, key):
return 'testuser'
def get(self, key, default=None):
return 'testuser'
class TestMainWindow(unittest.TestCase):
def setUp(self):
self.user = {'id': 1, 'username': 'testuser'}
# 避免实际弹窗和主循环
MainWindow.__init__ = lambda self, user: None
self.window = MainWindow(self.user)
def test_instance(self):
self.assertIsInstance(self.window, MainWindow)
def test_methods_exist(self):
self.assertTrue(hasattr(self.window, '_create_widgets'))
self.assertTrue(hasattr(self.window, '_analyze_content'))
self.assertTrue(hasattr(self.window, '_generate_strike_plan'))
if __name__ == '__main__':
unittest.main()

BIN
drone_decision_system/utils/__pycache__/database.cpython-313.pyc
View File

Binary file not shown.

BIN
drone_decision_system/utils/__pycache__/file_utils.cpython-313.pyc
View File

Binary file not shown.

305
drone_decision_system/utils/database.py
Unescape View File

@ -0,0 +1,305 @@
"""
无人机决策系统 - 数据库管理模块
作者：刘宇杰
日期：2025-07-13
功能说明：
本模块提供数据库的初始化、分析记录和打击计划的增删查改等操作。
"""
import sqlite3
import json
from sqlite3 import Error
from datetime import datetime
from config.settings import settings
from typing import List, Dict, Any, Optional
class DatabaseManager:
"""
数据库管理类，负责数据库的初始化和数据操作。
"""
def __init__(self):
"""
初始化数据库管理器，自动创建数据库和表结构。
"""
self.db_path = settings.DATABASE_PATH
self._init_db()
def _init_db(self):
"""
初始化数据库表结构。
"""
conn = None
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 简化后的分析记录表(不再关联用户)
cursor.execute("""
CREATE TABLE IF NOT EXISTS analysis_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
record_type TEXT NOT NULL, -- 'text' or 'image'
file_path TEXT NOT NULL,
analysis_result TEXT NOT NULL,
created_at TEXT NOT NULL
)
""")
# 简化后的打击计划表(不再关联用户)
cursor.execute("""
CREATE TABLE IF NOT EXISTS strike_plans (
id INTEGER PRIMARY KEY AUTOINCREMENT,
plan_details TEXT NOT NULL,
created_at TEXT NOT NULL,
executed BOOLEAN DEFAULT 0,
execution_result TEXT
)
""")
conn.commit()
except Error as e:
print(f"数据库初始化失败: {str(e)}")
finally:
if conn:
conn.close()
def add_analysis_record(self, record_type: str, file_path: str, analysis_result: dict) -> Optional[int]:
"""
添加分析记录。
:param record_type: 记录类型（text/image）
:param file_path: 文件路径
:param analysis_result: 分析结果字典
:return: 新记录ID或-1
"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(
"INSERT INTO analysis_records (record_type, file_path, analysis_result, created_at) VALUES (?, ?, ?, ?)",
(record_type, file_path, json.dumps(analysis_result), datetime.now().isoformat())
)
record_id = cursor.lastrowid
conn.commit()
return record_id
except Error as e:
print(f"添加分析记录失败: {str(e)}")
return -1
finally:
if conn:
conn.close()
def get_records(self, limit: int = 10) -> list:
"""
获取分析记录。
:param limit: 返回记录数
:return: 记录列表
"""
try:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM analysis_records ORDER BY created_at DESC LIMIT ?",
(limit,)
)
records = cursor.fetchall()
return [dict(record) for record in records]
except Error as e:
print(f"获取分析记录失败: {str(e)}")
return []
finally:
if conn:
conn.close()
def get_user_records(self, user_id: int, limit: int = 10) -> list:
"""
获取用户的分析记录（兼容旧接口）。
:param user_id: 用户ID
:param limit: 返回记录数
:return: 记录列表
"""
return self.get_records(limit)
def add_strike_plan(self, user_id: int, plan_details: dict) -> Optional[int]:
"""
添加打击计划。
:param user_id: 用户ID
:param plan_details: 计划详情字典
:return: 新计划ID或-1
"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(
"INSERT INTO strike_plans (plan_details, created_at) VALUES (?, ?)",
(json.dumps(plan_details), datetime.now().isoformat())
)
plan_id = cursor.lastrowid
conn.commit()
return plan_id
except Error as e:
print(f"添加打击计划失败: {str(e)}")
return -1
finally:
if conn:
conn.close()
def get_user_plans(self, user_id: int, limit: int = 5) -> list:
"""
获取用户的打击计划。
:param user_id: 用户ID
:param limit: 返回计划数
:return: 计划列表
"""
try:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM strike_plans ORDER BY created_at DESC LIMIT ?",
(limit,)
)
plans = cursor.fetchall()
return [dict(plan) for plan in plans]
except Error as e:
print(f"获取打击计划失败: {str(e)}")
return []
finally:
if conn:
conn.close()
def mark_plan_executed(self, plan_id: int, execution_result: dict):
"""
标记计划为已执行。
:param plan_id: 计划ID
:param execution_result: 执行结果字典
"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(
"UPDATE strike_plans SET executed = 1, execution_result = ? WHERE id = ?",
(json.dumps(execution_result), plan_id)
)
conn.commit()
except Error as e:
print(f"更新计划状态失败: {str(e)}")
finally:
if conn:
conn.close()
def get_user(self, username: str) -> dict:
"""
获取用户信息（模拟用户）。
:param username: 用户名
:return: 用户信息字典
"""
# 简化版：返回模拟用户数据
return {
'id': 1,
'username': username,
'password_hash': '', # 简化版不验证密码
'created_at': datetime.now().isoformat(),
'last_login': None
}
def add_user(self, username: str, password_hash: str) -> bool:
"""
添加新用户（模拟）。
:param username: 用户名
:param password_hash: 密码哈希
:return: 是否成功
"""
# 简化版：总是返回成功
print(f"添加用户: {username}")
return True
def update_user_login_time(self, user_id: int):
"""
更新用户最后登录时间（模拟）。
:param user_id: 用户ID
"""
# 简化版：不实际更新数据库
print(f"更新用户 {user_id} 登录时间")
def unused_function_3():
temp = [x * 2 for x in range(7)]
return temp
class UnusedClassC:
def __init__(self):
self.data = []
def add(self, item):
self.data.append(item)
def clear(self):
self.data = []
unused_var_3 = 3.14159
unused_set_3 = set(range(4))
def unused_function_11():
s = 0
for i in range(50):
s += i
return s
def unused_function_12():
return [chr(97+i) for i in range(26)]
class UnusedClassK:
def __init__(self):
self.items = set()
def add_item(self, item):
self.items.add(item)
def clear_items(self):
self.items.clear()
class UnusedClassL:
def __init__(self):
self.counter = 0
def inc(self):
self.counter += 1
def reset(self):
self.counter = 0
unused_var_14 = [i for i in range(100)]
unused_var_15 = 'database_unused'
unused_var_16 = {i: i*i for i in range(10)}
def unused_function_23():
s = 0
for i in range(200):
s += i
return s
def unused_function_24():
return [chr(65+i) for i in range(52)]
class UnusedClassW:
def __init__(self):
self.items = set(range(100))
def add_item(self, item):
self.items.add(item)
def clear_items(self):
self.items.clear()
class UnusedClassX:
def __init__(self):
self.counter = 100
def dec(self):
self.counter -= 1
def reset(self):
self.counter = 100
unused_var_33 = [i for i in range(200)]
unused_var_34 = 'database_more_unused'
unused_var_35 = {i: i*i for i in range(20)}

288
drone_decision_system/utils/file_utils.py
Unescape View File

@ -0,0 +1,288 @@
"""
无人机决策系统 - 文件工具模块
作者：刘宇杰
日期：2025-07-13
功能说明：
本文件提供文件操作相关的通用方法，包括安全保存、读取、校验、元数据获取等。
"""
import os
import hashlib
from datetime import datetime
from typing import Union, Optional, Tuple
from pathlib import Path
from config.settings import settings
import mimetypes
class FileUtils:
"""
文件操作工具类，提供文件处理相关的通用方法。
"""
@staticmethod
def is_allowed_file(filename: str, file_type: str = 'image') -> bool:
"""
检查文件扩展名是否允许。
参数：
filename (str): 文件名。
file_type (str): 文件类型('image'或'text')。
返回：
bool: 是否允许。
"""
if '.' not in filename:
return False
ext = filename.rsplit('.', 1)[1].lower()
if file_type == 'image':
return ext in settings.ALLOWED_IMAGE_EXTENSIONS
elif file_type == 'text':
return ext in settings.ALLOWED_TEXT_EXTENSIONS
else:
return False
@staticmethod
def generate_safe_filename(filename: str) -> str:
"""
生成安全的文件名（防止路径遍历和特殊字符）。
参数：
filename (str): 原始文件名。
返回：
str: 安全的文件名。
"""
filename = os.path.basename(filename)
safe_chars = "-_.() %s%s" % (os.path.sep, os.path.altsep) if os.path.altsep else "-_.() "
filename = ''.join(c for c in filename if c.isalnum() or c in safe_chars).strip()
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
name, ext = os.path.splitext(filename)
return f"{name}_{timestamp}{ext}"
@staticmethod
def get_file_hash(file_path: Union[str, Path], algorithm: str = 'sha256') -> Optional[str]:
"""
计算文件哈希值。
参数：
file_path (str|Path): 文件路径。
algorithm (str): 哈希算法('md5', 'sha1', 'sha256')。
返回：
str|None: 哈希值或None。
"""
hash_func = getattr(hashlib, algorithm, None)
if not hash_func:
return None
try:
with open(file_path, 'rb') as f:
file_hash = hash_func()
while chunk := f.read(8192):
file_hash.update(chunk)
return file_hash.hexdigest()
except IOError:
return None
@staticmethod
def create_secure_upload_dir(dir_path: Union[str, Path]) -> bool:
"""
创建安全的上传目录（设置适当权限）。
参数：
dir_path (str|Path): 目录路径。
返回：
bool: 是否成功。
"""
try:
os.makedirs(dir_path, exist_ok=True)
os.chmod(dir_path, 0o700)
return True
except OSError:
return False
@staticmethod
def save_file(file_data: bytes, save_path: Union[str, Path], overwrite: bool = False) -> bool:
"""
安全保存文件。
参数：
file_data (bytes): 文件二进制数据。
save_path (str|Path): 保存路径。
overwrite (bool): 是否覆盖已存在文件。
返回：
bool: 是否成功。
"""
if os.path.exists(save_path) and not overwrite:
return False
try:
with open(save_path, 'wb') as f:
f.write(file_data)
os.chmod(save_path, 0o600)
return True
except IOError:
return False
@staticmethod
def read_file(file_path: Union[str, Path], mode: str = 'rb') -> Optional[Union[bytes, str]]:
"""
安全读取文件。
参数：
file_path (str|Path): 文件路径。
mode (str): 读取模式('rb'或'r')。
返回：
bytes|str|None: 文件内容或None。
"""
if not os.path.exists(file_path):
return None
try:
with open(file_path, mode) as f:
return f.read()
except IOError:
return None
@staticmethod
def get_file_metadata(file_path: Union[str, Path]) -> Optional[dict]:
"""
获取文件元数据（大小、类型、修改时间等）。
参数：
file_path (str|Path): 文件路径。
返回：
dict|None: 元数据字典或None。
"""
if not os.path.exists(file_path):
return None
try:
stat = os.stat(file_path)
mime_type, _ = mimetypes.guess_type(file_path)
return {
'size': stat.st_size,
'mime_type': mime_type or 'application/octet-stream',
'created_at': datetime.fromtimestamp(stat.st_ctime).isoformat(),
'modified_at': datetime.fromtimestamp(stat.st_mtime).isoformat(),
'accessed_at': datetime.fromtimestamp(stat.st_atime).isoformat()
}
except OSError:
return None
@staticmethod
def split_file_path(file_path: Union[str, Path]) -> Tuple[str, str, str]:
"""
拆分文件路径为目录、文件名和扩展名。
参数：
file_path (str|Path): 文件路径。
返回：
(str, str, str): (目录, 文件名, 扩展名)
"""
dir_path = os.path.dirname(file_path)
filename = os.path.basename(file_path)
name, ext = os.path.splitext(filename)
return dir_path, name, ext
@staticmethod
def get_unique_filename(directory: Union[str, Path], filename: str) -> str:
"""
获取唯一的文件名（避免覆盖）。
参数：
directory (str|Path): 目录路径。
filename (str): 原始文件名。
返回：
str: 唯一文件名。
"""
base, ext = os.path.splitext(filename)
counter = 1
new_filename = filename
while os.path.exists(os.path.join(directory, new_filename)):
new_filename = f"{base}_{counter}{ext}"
counter += 1
return new_filename
@staticmethod
def validate_file_content(file_path: Union[str, Path], expected_type: Optional[str] = None) -> bool:
"""
验证文件内容是否符合预期类型（简单验证）。
参数：
file_path (str|Path): 文件路径。
expected_type (str|None): 预期类型('image'或'text')。
返回：
bool: 是否验证通过。
"""
if not os.path.exists(file_path):
return False
try:
with open(file_path, 'rb') as f:
header = f.read(32)
if expected_type == 'image':
image_magic_numbers = {
b'\xFF\xD8\xFF': 'jpg',
b'\x89PNG': 'png',
b'GIF87a': 'gif',
b'GIF89a': 'gif',
b'BM': 'bmp'
}
return any(header.startswith(magic) for magic in image_magic_numbers)
elif expected_type == 'text':
try:
header.decode('utf-8')
return True
except UnicodeDecodeError:
return False
else:
return True
except IOError:
return False
def unused_function_4():
for _ in range(2):
pass
return '无用函数4'
class UnusedClassD:
def __init__(self):
self.flag = False
def toggle(self):
self.flag = not self.flag
unused_var_4 = [1,2,3,4]
unused_tuple_4 = (None, None)
def unused_function_13():
return sum([i for i in range(30)])
def unused_function_14():
s = ''
for i in range(15):
s += chr(65+i)
return s
class UnusedClassM:
def __init__(self):
self.values = [0]*5
def set_value(self, idx, val):
if 0 <= idx < 5:
self.values[idx] = val
def get_values(self):
return self.values
class UnusedClassN:
def __init__(self):
self.active = False
def activate(self):
self.active = True
def deactivate(self):
self.active = False
unused_var_17 = [i**2 for i in range(20)]
unused_var_18 = 'file_utils_unused'
unused_var_19 = (None, 1, 2, 3)
def unused_function_25():
return sum([i for i in range(100)])
def unused_function_26():
s = ''
for i in range(100):
s += chr(97 + (i % 26))
return s
class UnusedClassY:
def __init__(self):
self.values = [0]*20
def set_value(self, idx, val):
if 0 <= idx < 20:
self.values[idx] = val
def get_values(self):
return self.values
class UnusedClassZ:
def __init__(self):
self.active = False
def activate(self):
self.active = True
def deactivate(self):
self.active = False
unused_var_36 = [i**2 for i in range(100)]
unused_var_37 = 'file_utils_more_unused'
unused_var_38 = (None, 1, 2, 3, 4, 5)
Write Preview
Loading…
Cancel
Save