ZZH000829/archery
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

upload log file to qiqiu

Browse Source
This commit is contained in:
gcw_4spBpAfv
2026-04-23 17:53:21 +08:00
parent 12fac4ea1c
commit 8efe1ae5c5
5 changed files with 316 additions and 877 deletions
Download Patch File Download Diff File Expand all files Collapse all files

363
network.py
View File

@@ -1358,6 +1358,237 @@ class NetworkManager:
self.logger.error(f"[LOG_UPLOAD] 上传异常: {e}")
self.safe_enqueue({"result": "log_upload_failed", "reason": str(e)[:100]}, 2)
def _prepare_log_archive(self, include_rotated=True, max_files=None, archive_format="tgz"):
"""准备日志归档压缩包，返回 (archive_path, archive_filename) 或 (None, error_msg)
Args:
include_rotated: 是否包含轮转日志
max_files: 最多打包多少个日志文件
archive_format: tgz 或 zip
"""
import shutil
from datetime import datetime
import glob
try:
log_file_path = config.LOG_FILE
if not os.path.exists(log_file_path):
return None, "log_file_not_found"
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
device_id = self._device_id or "unknown"
base_name = f"logs_{timestamp}_{device_id}"
archive_format = (archive_format or "tgz").strip().lower()
if archive_format not in ("tgz", "zip"):
archive_format = "tgz"
candidates = [log_file_path]
if include_rotated:
candidates = sorted(set(glob.glob(log_file_path + "*")))
candidates = [p for p in candidates if os.path.isfile(p)]
def _log_sort_key(p):
if p == log_file_path:
return (0, 0, p)
suffix = p[len(log_file_path):]
if suffix.startswith("."):
try:
return (1, int(suffix[1:]), p)
except:
return (2, 999999, p)
return (3, 999999, p)
candidates.sort(key=_log_sort_key)
if max_files is None:
try:
max_files = 1 + int(getattr(config, "LOG_BACKUP_COUNT", 5))
except:
max_files = 6
try:
max_files = int(max_files)
except:
max_files = 6
max_files = max(1, min(max_files, 20))
selected = candidates[:max_files]
if not selected:
return None, "no_log_files"
os.system("sync")
temp_dir = "/tmp"
staging_dir = os.path.join(temp_dir, f"log_upload_{base_name}")
os.makedirs(staging_dir, exist_ok=True)
staged_paths = []
try:
for p in selected:
dst = os.path.join(staging_dir, os.path.basename(p))
shutil.copy2(p, dst)
staged_paths.append(dst)
except Exception as e:
try:
shutil.rmtree(staging_dir)
except:
pass
return None, f"snapshot_failed: {e}"
if archive_format == "zip":
archive_filename = f"{base_name}.zip"
else:
archive_filename = f"{base_name}.tar.gz"
archive_path = os.path.join(temp_dir, archive_filename)
try:
if archive_format == "zip":
import zipfile
with zipfile.ZipFile(archive_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
for p in staged_paths:
zf.write(p, arcname=os.path.basename(p))
else:
import tarfile
with tarfile.open(archive_path, "w:gz") as tf:
for p in staged_paths:
tf.add(p, arcname=os.path.basename(p))
except Exception as e:
try:
shutil.rmtree(staging_dir)
except:
pass
try:
if os.path.exists(archive_path):
os.remove(archive_path)
except:
pass
return None, f"archive_failed: {e}"
finally:
try:
shutil.rmtree(staging_dir)
except:
pass
return archive_path, archive_filename
except Exception as e:
return None, f"prepare_exception: {e}"
def _upload_log_file_v2(self, upload_url, upload_token, key, outlink="", include_rotated=True, max_files=None, archive_format="tgz"):
"""上传日志到 Qiniu（支持 WiFi 和 4G 双路径）
流程：准备日志归档 -> 自动检测网络 -> WiFi(requests) 或 4G(AT命令) 上传
"""
import shutil
# 1) 准备日志归档
archive_path, info = self._prepare_log_archive(include_rotated, max_files, archive_format)
if archive_path is None:
self.logger.error(f"[LOG_UPLOAD] 准备归档失败: {info}")
self.safe_enqueue({"result": "log_upload_failed", "reason": info}, 2)
return
archive_filename = info
# key 是服务器下发的目录前缀，最终 key = prefix/filename
qiniu_key = key.rstrip("/") + "/" + archive_filename
self.logger.info(f"[LOG_UPLOAD] 日志归档已生成: {archive_path}, qiniu_key: {qiniu_key}")
try:
# 2) WiFi 优先：只要 WiFi 已连接就先尝试 WiFi，失败再回落到 4G
wifi_tried = False
wifi_ok = False
if self.is_wifi_connected():
wifi_tried = True
self.logger.info(f"[LOG_UPLOAD] Using wifi path (preferred), archive: {archive_path}")
try:
# ---- WiFi path: 使用 requests 库上传 ----
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
with open(archive_path, 'rb') as f:
files = {'file': (archive_filename, f, 'application/octet-stream')}
data = {'token': upload_token, 'key': qiniu_key}
wifi_upload_url = upload_url.replace('https://', 'http://', 1)
self.logger.info(f"[LOG_UPLOAD] WiFi upload URL: {wifi_upload_url}")
response = requests.post(wifi_upload_url, files=files, data=data, timeout=120, verify=False)
response.raise_for_status()
result_json = response.json()
uploaded_key = result_json.get('key', qiniu_key)
self.logger.info(f"[LOG_UPLOAD] WiFi upload ok: key={uploaded_key}")
access_url = None
if outlink:
access_url = f"https://{outlink}/{uploaded_key}"
response_data = {
"result": "log_upload_ok",
"key": uploaded_key,
"via": "wifi",
}
if access_url:
response_data["url"] = access_url
self.safe_enqueue(response_data, 2)
wifi_ok = True
except Exception as e:
# WiFi 上传失败不影响主链路：记录原因并回落 4G
self.logger.warning(f"[LOG_UPLOAD] WiFi upload failed, fallback to 4g: {e}")
if not wifi_ok:
if not wifi_tried:
self.logger.info(f"[LOG_UPLOAD] WiFi not connected, using 4g path, archive: {archive_path}")
else:
self.logger.info(f"[LOG_UPLOAD] Using 4g fallback path, archive: {archive_path}")
# ---- 4G path: 使用 FourGUploadManager AT命令上传 ----
import importlib.util
spec = importlib.util.spec_from_file_location(
"four_g_upload_manager",
os.path.join(os.path.dirname(__file__), "4g_upload_manager.py")
)
upload_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(upload_module)
FourGUploadManager = upload_module.FourGUploadManager
uploader = FourGUploadManager(hardware_manager.at_client)
result = uploader.upload_file(archive_path, upload_url, upload_token, qiniu_key)
if result.get("success"):
uploaded_key = result.get("key", qiniu_key)
self.logger.info(f"[LOG_UPLOAD] 4G upload ok: key={uploaded_key}")
access_url = None
if outlink:
access_url = f"https://{outlink}/{uploaded_key}"
response_data = {
"result": "log_upload_ok",
"key": uploaded_key,
"via": "4g",
}
if access_url:
response_data["url"] = access_url
self.safe_enqueue(response_data, 2)
else:
error_msg = result.get("error", "unknown_error")
self.logger.error(f"[LOG_UPLOAD] 4G upload failed: {error_msg}")
self.safe_enqueue({
"result": "log_upload_failed",
"reason": error_msg[:100]
}, 2)
except Exception as e:
self.logger.error(f"[LOG_UPLOAD] upload exception: {e}")
self.safe_enqueue({"result": "log_upload_failed", "reason": str(e)[:100]}, 2)
finally:
# 清理临时归档文件
try:
if archive_path and os.path.exists(archive_path):
os.remove(archive_path)
self.logger.debug(f"[LOG_UPLOAD] 临时归档已删除: {archive_path}")
except Exception as e:
self.logger.warning(f"[LOG_UPLOAD] 删除临时归档失败: {e}")
def _upload_image_file(self, image_path, upload_url, upload_token, key, shoot_id, outlink):
"""上传图片文件到指定URL（自动检测网络类型，WiFi使用requests，4G使用AT HTTP命令）
@@ -1369,51 +1600,57 @@ class NetworkManager:
shoot_id: 射击ID
outlink: 外链域名（可选，用于构建访问URL）
"""
# 自动检测网络类型，选择上传路径
if self._network_type == "wifi" and self.is_wifi_connected():
mode = "wifi"
else:
mode = "4g"
# WiFi 优先（独立于 TCP 主链路）：只要 WiFi 已连接就先走 WiFi，失败再回落 4G
mode = "wifi" if self.is_wifi_connected() else "4g"
self.logger.info(f"[IMAGE_UPLOAD] Using {mode} path, image: {image_path}")
try:
wifi_ok = False
if mode == "wifi":
# ---- WiFi path: 使用 requests 库上传 ----
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
try:
# ---- WiFi path: 使用 requests 库上传 ----
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
with open(image_path, 'rb') as f:
files = {'file': (os.path.basename(image_path), f, 'application/octet-stream')}
data = {'token': upload_token, 'key': key}
# 测试：将HTTPS转为HTTP
wifi_upload_url = upload_url.replace('https://', 'http://', 1)
self.logger.info(f"[IMAGE_UPLOAD] WiFi upload URL: {wifi_upload_url}")
response = requests.post(wifi_upload_url, files=files, data=data, timeout=120, verify=False)
response.raise_for_status()
result_json = response.json()
uploaded_key = result_json.get('key', key)
with open(image_path, 'rb') as f:
files = {'file': (os.path.basename(image_path), f, 'application/octet-stream')}
data = {'token': upload_token, 'key': key}
# 将 HTTPS 转为 HTTP（设备端 SSL 兼容性）
wifi_upload_url = upload_url.replace('https://', 'http://', 1)
self.logger.info(f"[IMAGE_UPLOAD] WiFi upload URL: {wifi_upload_url}")
response = requests.post(wifi_upload_url, files=files, data=data, timeout=120, verify=False)
response.raise_for_status()
result_json = response.json()
uploaded_key = result_json.get('key', key)
self.logger.info(f"[IMAGE_UPLOAD] WiFi upload ok: key={uploaded_key}")
self.logger.info(f"[IMAGE_UPLOAD] WiFi upload ok: key={uploaded_key}")
access_url = None
if outlink:
access_url = f"https://{outlink}/{uploaded_key}"
access_url = None
if outlink:
access_url = f"https://{outlink}/{uploaded_key}"
response_data = {
"result": "image_upload_ok",
"shootId": shoot_id,
"key": uploaded_key,
"via": "wifi",
}
if access_url:
response_data["url"] = access_url
response_data = {
"result": "image_upload_ok",
"shootId": shoot_id,
"key": uploaded_key,
"via": "wifi",
}
if access_url:
response_data["url"] = access_url
self.safe_enqueue(response_data, 2)
self.safe_enqueue(response_data, 2)
wifi_ok = True
except Exception as e:
self.logger.warning(f"[IMAGE_UPLOAD] WiFi upload failed, fallback to 4g: {e}")
else:
if not wifi_ok:
# ---- 4G path: 使用 FourGUploadManager AT命令上传 ----
if mode != "4g":
self.logger.info(f"[IMAGE_UPLOAD] Using 4g fallback path, image: {image_path}")
import importlib.util
spec = importlib.util.spec_from_file_location(
"four_g_upload_manager",
@@ -1702,6 +1939,35 @@ class NetworkManager:
)
# 立即返回已入队确认
self.safe_enqueue({"result": "image_upload_queued", "shootId": shoot_id}, 2)
elif logged_in and msg_type == 101:
self.logger.info(f"[LOG_UPLOAD] 收到日志上传命令 {body}")
if isinstance(body, dict):
upload_url = body.get("uploadUrl")
upload_token = body.get("token")
key = body.get("key")
outlink = body.get("outlink", "")
include_rotated = body.get("includeRotated", True)
max_files = body.get("maxFiles")
archive_format = body.get("archive", "tgz")
hardware_manager.start_idle_timer() # 重新计时
# 验证必需字段
if not upload_url or not upload_token or not key:
self.logger.error("[LOG_UPLOAD] 缺少必需参数: uploadUrl, token 或 key")
self.safe_enqueue({"result": "log_upload_failed", "reason": "missing_params"}, 2)
else:
self.logger.info(f"[LOG_UPLOAD] 收到日志上传命令，key: {key}")
# 在新线程中执行上传，避免阻塞主循环
import _thread
_thread.start_new_thread(
self._upload_log_file_v2,
(upload_url, upload_token, key, outlink, include_rotated, max_files, archive_format)
)
# 立即返回已入队确认
self.safe_enqueue({"result": "log_upload_queued"}, 2)
# 处理业务指令
elif logged_in and isinstance(body, dict):
inner_cmd = None
@@ -1795,29 +2061,6 @@ class NetworkManager:
mccid = self.get_4g_mccid()
self.logger.info(f"4G MCCID: {mccid}")
self.safe_enqueue({"result": "mccid", "mccid": mccid if mccid is not None else ""}, 2)
# elif inner_cmd == 7:
# from ota_manager import ota_manager
# if ota_manager.update_thread_started:
# self.safe_enqueue({"result": "update_already_started"}, 2)
# continue
# try:
# ip = os.popen("ifconfig wlan0 2>/dev/null | grep 'inet ' | awk '{print $2}'").read().strip()
# except:
# ip = None
# if not ip:
# self.safe_enqueue({"result": "ota_rejected", "reason": "no_wifi_ip"}, 2)
# else:
# # 注意：direct_ota_download 需要 ota_url 参数
# # 如果 ota_manager.ota_url 为 None，需要从其他地方获取
# ota_url_to_use = ota_manager.ota_url
# if not ota_url_to_use:
# self.logger.error("[OTA] cmd=7 但 OTA_URL 未设置")
# self.safe_enqueue({"result": "ota_failed", "reason": "ota_url_not_set"}, 2)
# else:
# ota_manager._start_update_thread()
# _thread.start_new_thread(ota_manager.direct_ota_download, (ota_url_to_use,))
elif inner_cmd == 41:
self.logger.info(f"[TEST] 收到TCP射箭触发命令, {time.time()}")
self._manual_trigger_flag = True
@@ -1953,10 +2196,10 @@ class NetworkManager:
except Exception as e:
self.logger.error(f"[OTA] 检查 pending 文件时出错: {e}")
# 心跳超时重连
if logged_in and current_time - last_heartbeat_ack_time > 1000*60*10:
self.logger.error("十分钟无心跳ACK，重连")
break
# 服务器不再发送心跳ACK
# if logged_in and current_time - last_heartbeat_ack_time > 1000*60*10:
# self.logger.error("十分钟无心跳ACK，重连")
# break
self._send_event.wait(timeout=0.05) # 0.05秒 = 50ms
self._send_event.clear()