dev/seabox_server
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add tp_backup

Browse Source
This commit is contained in:
unknown
2025-08-29 21:15:06 +08:00
parent cf8e046b01
commit f574ecf3ae
31 changed files with 3304 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
backend/_oneoff_fix_all_migrations.py
View File

@ -1,25 +0,0 @@
import os
import sys
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
try:
import django
django.setup()
except Exception as e:
print(f"[ERROR] Django setup failed: {e}")
sys.exit(1)
from django.db import connection
try:
with connection.cursor() as c:
# Clear all migration history to start fresh
c.execute("DELETE FROM django_migrations")
deleted = c.rowcount
connection.commit()
print(f"[OK] Cleared all migration history: {deleted} rows deleted")
sys.exit(0)
except Exception as e:
print(f"[ERROR] Failed to clear migration history: {e}")
sys.exit(2)

28
backend/_oneoff_fix_auth.py
View File

@ -1,28 +0,0 @@
import os
import sys
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
try:
import django
django.setup()
except Exception as e:
print(f"[ERROR] Django setup failed: {e}")
sys.exit(1)
from django.db import connection
try:
with connection.cursor() as c:
c.execute(
"DELETE FROM django_migrations WHERE app=%s AND name=%s",
["auth", "0001_initial"],
)
deleted = c.rowcount
connection.commit()
print(f"[OK] Deleted rows in django_migrations for auth: {deleted}")
sys.exit(0)
except Exception as e:
print(f"[ERROR] Failed to fix auth migration history: {e}")
sys.exit(2)

42
backend/_oneoff_fix_auth_many.py
View File

@ -1,42 +0,0 @@
import os
import sys
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
try:
import django
django.setup()
except Exception as e:
print(f"[ERROR] Django setup failed: {e}")
sys.exit(1)
from django.db import connection
NAMES = (
'0002_alter_permission_name_max_length',
'0003_alter_user_email_max_length',
'0004_alter_user_username_opts',
'0005_alter_user_last_login_null',
'0006_require_contenttypes_0002',
'0007_alter_validators_add_error_messages',
'0008_alter_user_username_max_length',
'0009_alter_user_last_name_max_length',
'0010_alter_group_name_max_length',
'0011_update_proxy_permissions',
'0012_alter_user_first_name_max_length',
)
try:
with connection.cursor() as c:
c.executemany(
"DELETE FROM django_migrations WHERE app=%s AND name=%s",
[("auth", n) for n in NAMES],
)
deleted = c.rowcount
connection.commit()
print(f"[OK] Deleted rows for auth:* : {deleted}")
sys.exit(0)
except Exception as e:
print(f"[ERROR] Failed to batch delete auth migrations: {e}")
sys.exit(2)

29
backend/_oneoff_fix_contenttypes.py
View File

@ -1,29 +0,0 @@
import os
import sys
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
try:
import django
django.setup()
except Exception as e:
print(f"[ERROR] Django setup failed: {e}")
sys.exit(1)
from django.db import connection
try:
with connection.cursor() as c:
# Remove the incorrect history row if it exists
c.execute(
"DELETE FROM django_migrations WHERE app=%s AND name=%s",
["contenttypes", "0002_remove_content_type_name"],
)
deleted = c.rowcount
connection.commit()
print(f"[OK] Deleted rows in django_migrations: {deleted}")
sys.exit(0)
except Exception as e:
print(f"[ERROR] Failed to fix migration history: {e}")
sys.exit(2)

33
backend/_oneoff_fix_contenttypes_table.py
View File

@ -1,33 +0,0 @@
import os
import sys
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
try:
import django
django.setup()
except Exception as e:
print(f"[ERROR] Django setup failed: {e}")
sys.exit(1)
from django.db import connection
try:
with connection.cursor() as c:
# Check if name column exists
c.execute("SHOW COLUMNS FROM django_content_type LIKE 'name'")
has_name = c.fetchone()
if has_name:
print("[INFO] 'name' column exists, removing it...")
c.execute("ALTER TABLE django_content_type DROP COLUMN name")
print("[OK] Removed 'name' column from django_content_type")
else:
print("[INFO] 'name' column does not exist, table is already correct")
connection.commit()
print("[OK] django_content_type table structure fixed")
sys.exit(0)
except Exception as e:
print(f"[ERROR] Failed to fix django_content_type table: {e}")
sys.exit(2)

1
backend/application/settings.py
View File

@ -65,6 +65,7 @@ INSTALLED_APPS = [
#主要添加如下代码
My_Apps = [
'translate', #新的应用写在这里
'session', #会话管理应用
]
INSTALLED_APPS += My_Apps

4
backend/application/urls.py
View File

@ -123,7 +123,9 @@ urlpatterns = (
#就是添加如下内容，把自己的路由单独写出来，这样方便与dvadmin3的官方路由作区分
My_Urls = (
[ #这里的crud_demo是指django创建的应用名称crud_demo
path('',include('translate.urls')),]
path('',include('translate.urls')),
path('',include('session.urls')), # 会话管理路由
]
)
# 这里把自己的路径单独出来，后面再追加在一起

65
backend/check_tables.py Normal file
View File

@ -0,0 +1,65 @@
#!/usr/bin/env python3
import os
import django
# 设置Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
django.setup()
from django.db import connection
def check_table_structure():
"""检查关键表的结构"""
cursor = connection.cursor()
# 检查的表列表
tables_to_check = [
'language_list',
'translate_route',
'translate_config',
'session_list',
'contact_info',
'translate_cache',
'proxy_config',
'global_proxy_config',
'group_manage',
'quick_reply_record',
'parameter',
'tg_sessions',
'follow_record'
]
for table_name in tables_to_check:
print(f"\n=== {table_name} 表结构 ===")
try:
cursor.execute(f'DESCRIBE {table_name}')
fields = cursor.fetchall()
print(f"{'字段名':<20} {'类型':<25} {'空值':<5} {'键':<5} {'默认值':<15}")
print("-" * 75)
for field in fields:
field_name = field[0]
field_type = field[1]
null_allowed = field[2]
key_type = field[3]
default_value = field[4] if field[4] is not None else 'NULL'
print(f"{field_name:<20} {field_type:<25} {null_allowed:<5} {key_type:<5} {default_value:<15}")
except Exception as e:
print(f"错误: {e}")
# 检查数据量
print(f"\n=== 数据量统计 ===")
for table_name in tables_to_check:
try:
cursor.execute(f'SELECT COUNT(*) FROM {table_name}')
count = cursor.fetchone()[0]
print(f"{table_name:<20}: {count:>6} 条记录")
except Exception as e:
print(f"{table_name:<20}: 错误 - {e}")
if __name__ == "__main__":
check_table_structure()

328
backend/complete_mysql_import.sql Normal file
View File

@ -0,0 +1,328 @@
-- 完整的 MySQL 数据库导入文件
-- 包含所有 SQLite 表的 MySQL 版本
USE fyapi;
-- 设置字符集
SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
-- ================================
-- 会话管理相关表
-- ================================
-- 会话列表表
DROP TABLE IF EXISTS `session_list`;
CREATE TABLE `session_list` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`partitionId` varchar(255) NOT NULL COMMENT '分区ID',
`windowStatus` varchar(10) DEFAULT 'false' COMMENT '窗口状态',
`createTime` varchar(50) DEFAULT NULL COMMENT '创建时间',
`platform` varchar(100) DEFAULT NULL COMMENT '平台',
`nickName` varchar(255) DEFAULT NULL COMMENT '昵称',
`msgCount` int(11) DEFAULT 0 COMMENT '消息数量',
`onlineStatus` varchar(10) DEFAULT 'false' COMMENT '在线状态',
`webUrl` text COMMENT '网页URL',
`windowId` int(11) DEFAULT 0 COMMENT '窗口ID',
`userAgent` text COMMENT '用户代理',
`remarks` text COMMENT '备注',
`avatarUrl` text COMMENT '头像URL',
`userName` varchar(255) DEFAULT NULL COMMENT '用户名',
`isTop` varchar(10) DEFAULT 'false' COMMENT '是否置顶',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
UNIQUE KEY `idx_partition_id` (`partitionId`),
KEY `idx_platform` (`platform`),
KEY `idx_create_time` (`createTime`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='会话列表';
-- 联系人信息表
DROP TABLE IF EXISTS `contact_info`;
CREATE TABLE `contact_info` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`userId` varchar(255) NOT NULL COMMENT '用户ID',
`platform` varchar(100) NOT NULL COMMENT '平台',
`nickName` varchar(255) DEFAULT NULL COMMENT '昵称',
`userName` varchar(255) DEFAULT NULL COMMENT '用户名',
`avatarUrl` text COMMENT '头像URL',
`phoneNumber` varchar(50) DEFAULT NULL COMMENT '电话号码',
`email` varchar(255) DEFAULT NULL COMMENT '邮箱',
`remarks` text COMMENT '备注',
`level` varchar(50) DEFAULT NULL COMMENT '等级',
`tags` text COMMENT '标签',
`lastContactTime` varchar(50) DEFAULT NULL COMMENT '最后联系时间',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
UNIQUE KEY `idx_user_platform` (`userId`, `platform`),
KEY `idx_platform` (`platform`),
KEY `idx_nick_name` (`nickName`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='联系人信息';
-- 跟进记录表
DROP TABLE IF EXISTS `follow_record`;
CREATE TABLE `follow_record` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`userId` varchar(255) NOT NULL COMMENT '用户ID',
`platform` varchar(100) NOT NULL COMMENT '平台',
`content` text COMMENT '跟进内容',
`followTime` varchar(50) DEFAULT NULL COMMENT '跟进时间',
`followType` varchar(50) DEFAULT NULL COMMENT '跟进类型',
`result` varchar(255) DEFAULT NULL COMMENT '跟进结果',
`nextFollowTime` varchar(50) DEFAULT NULL COMMENT '下次跟进时间',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
KEY `idx_user_platform` (`userId`, `platform`),
KEY `idx_follow_time` (`followTime`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='跟进记录';
-- ================================
-- 翻译相关表
-- ================================
-- 翻译配置表
DROP TABLE IF EXISTS `translate_config`;
CREATE TABLE `translate_config` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`userId` varchar(255) DEFAULT NULL COMMENT '用户ID',
`platform` varchar(100) DEFAULT NULL COMMENT '平台',
`sourceLanguage` varchar(50) DEFAULT 'auto' COMMENT '源语言',
`targetLanguage` varchar(50) DEFAULT 'zh' COMMENT '目标语言',
`translateService` varchar(100) DEFAULT 'google' COMMENT '翻译服务',
`autoTranslate` varchar(10) DEFAULT 'false' COMMENT '自动翻译',
`showOriginal` varchar(10) DEFAULT 'true' COMMENT '显示原文',
`translateDelay` int(11) DEFAULT 1000 COMMENT '翻译延迟(毫秒)',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
KEY `idx_user_platform` (`userId`, `platform`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='翻译配置';
-- 翻译路由表
DROP TABLE IF EXISTS `translate_route`;
CREATE TABLE `translate_route` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`sourceLang` varchar(50) NOT NULL COMMENT '源语言',
`targetLang` varchar(50) NOT NULL COMMENT '目标语言',
`translateService` varchar(100) NOT NULL COMMENT '翻译服务',
`priority` int(11) DEFAULT 1 COMMENT '优先级',
`isActive` varchar(10) DEFAULT 'true' COMMENT '是否激活',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
KEY `idx_lang_pair` (`sourceLang`, `targetLang`),
KEY `idx_service` (`translateService`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='翻译路由';
-- 语言列表表
DROP TABLE IF EXISTS `language_list`;
CREATE TABLE `language_list` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`languageCode` varchar(50) NOT NULL COMMENT '语言代码',
`languageName` varchar(255) NOT NULL COMMENT '语言名称',
`nativeName` varchar(255) DEFAULT NULL COMMENT '本地名称',
`isActive` varchar(10) DEFAULT 'true' COMMENT '是否激活',
`sortOrder` int(11) DEFAULT 0 COMMENT '排序',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
UNIQUE KEY `idx_language_code` (`languageCode`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='语言列表';
-- 翻译缓存表
DROP TABLE IF EXISTS `translate_cache`;
CREATE TABLE `translate_cache` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`partitionId` varchar(255) DEFAULT NULL COMMENT '分区ID',
`sourceText` text COMMENT '原文',
`translatedText` text COMMENT '译文',
`sourceLang` varchar(50) DEFAULT NULL COMMENT '源语言',
`targetLang` varchar(50) DEFAULT NULL COMMENT '目标语言',
`translateService` varchar(100) DEFAULT NULL COMMENT '翻译服务',
`cacheTime` varchar(50) DEFAULT NULL COMMENT '缓存时间',
`hitCount` int(11) DEFAULT 1 COMMENT '命中次数',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
KEY `idx_partition_id` (`partitionId`),
KEY `idx_source_text` (`sourceText`(255)),
KEY `idx_lang_pair` (`sourceLang`, `targetLang`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='翻译缓存';
-- ================================
-- 代理配置相关表
-- ================================
-- 代理配置表
DROP TABLE IF EXISTS `proxy_config`;
CREATE TABLE `proxy_config` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`partitionId` varchar(255) NOT NULL COMMENT '分区ID',
`proxyStatus` varchar(10) DEFAULT 'false' COMMENT '代理状态',
`proxyType` varchar(50) DEFAULT 'http' COMMENT '代理类型',
`proxyIp` varchar(255) DEFAULT NULL COMMENT '代理IP',
`proxyPort` varchar(10) DEFAULT NULL COMMENT '代理端口',
`userVerifyStatus` varchar(10) DEFAULT 'false' COMMENT '用户验证状态',
`username` varchar(255) DEFAULT NULL COMMENT '用户名',
`password` varchar(255) DEFAULT NULL COMMENT '密码',
`timezone` varchar(100) DEFAULT NULL COMMENT '时区',
`defaultLanguage` varchar(50) DEFAULT 'zh-CN' COMMENT '默认语言',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
UNIQUE KEY `idx_partition_id` (`partitionId`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='代理配置';
-- 全局代理配置表
DROP TABLE IF EXISTS `global_proxy_config`;
CREATE TABLE `global_proxy_config` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`proxyStatus` varchar(10) DEFAULT 'false' COMMENT '代理状态',
`proxyType` varchar(50) DEFAULT 'http' COMMENT '代理类型',
`proxyIp` varchar(255) DEFAULT NULL COMMENT '代理IP',
`proxyPort` varchar(10) DEFAULT NULL COMMENT '代理端口',
`userVerifyStatus` varchar(10) DEFAULT 'false' COMMENT '用户验证状态',
`username` varchar(255) DEFAULT NULL COMMENT '用户名',
`password` varchar(255) DEFAULT NULL COMMENT '密码',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='全局代理配置';
-- ================================
-- 快捷回复相关表
-- ================================
-- 分组管理表
DROP TABLE IF EXISTS `group_manage`;
CREATE TABLE `group_manage` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`groupName` varchar(255) NOT NULL COMMENT '分组名称',
`groupDescription` text COMMENT '分组描述',
`sortOrder` int(11) DEFAULT 0 COMMENT '排序',
`isActive` varchar(10) DEFAULT 'true' COMMENT '是否激活',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
KEY `idx_group_name` (`groupName`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='分组管理';
-- 快捷回复记录表
DROP TABLE IF EXISTS `quick_reply_record`;
CREATE TABLE `quick_reply_record` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`groupId` int(11) DEFAULT NULL COMMENT '分组ID',
`title` varchar(255) NOT NULL COMMENT '标题',
`content` text NOT NULL COMMENT '内容',
`tags` varchar(500) DEFAULT NULL COMMENT '标签',
`useCount` int(11) DEFAULT 0 COMMENT '使用次数',
`sortOrder` int(11) DEFAULT 0 COMMENT '排序',
`isActive` varchar(10) DEFAULT 'true' COMMENT '是否激活',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
KEY `idx_group_id` (`groupId`),
KEY `idx_title` (`title`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='快捷回复记录';
-- 用户快捷回复表
DROP TABLE IF EXISTS `user_quick_replies`;
CREATE TABLE `user_quick_replies` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`userId` varchar(255) NOT NULL COMMENT '用户ID',
`title` varchar(255) NOT NULL COMMENT '标题',
`content` text NOT NULL COMMENT '内容',
`category` varchar(100) DEFAULT NULL COMMENT '分类',
`tags` varchar(500) DEFAULT NULL COMMENT '标签',
`useCount` int(11) DEFAULT 0 COMMENT '使用次数',
`isActive` varchar(10) DEFAULT 'true' COMMENT '是否激活',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
KEY `idx_user_id` (`userId`),
KEY `idx_category` (`category`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户快捷回复';
-- ================================
-- 系统配置相关表
-- ================================
-- 参数配置表
DROP TABLE IF EXISTS `parameter`;
CREATE TABLE `parameter` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`key` varchar(255) NOT NULL COMMENT '参数键',
`value` text COMMENT '参数值',
`description` varchar(500) DEFAULT NULL COMMENT '描述',
`type` varchar(50) DEFAULT 'string' COMMENT '类型',
`isSystem` varchar(10) DEFAULT 'false' COMMENT '是否系统参数',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
UNIQUE KEY `idx_key` (`key`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='参数配置';
-- Telegram 会话表
DROP TABLE IF EXISTS `tg_sessions`;
CREATE TABLE `tg_sessions` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`sessionId` varchar(255) NOT NULL COMMENT '会话ID',
`userId` varchar(255) DEFAULT NULL COMMENT '用户ID',
`phoneNumber` varchar(50) DEFAULT NULL COMMENT '电话号码',
`sessionData` longtext COMMENT '会话数据',
`isActive` varchar(10) DEFAULT 'true' COMMENT '是否激活',
`lastUsed` varchar(50) DEFAULT NULL COMMENT '最后使用时间',
`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`),
UNIQUE KEY `idx_session_id` (`sessionId`),
KEY `idx_user_id` (`userId`),
KEY `idx_phone_number` (`phoneNumber`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Telegram 会话';
-- ================================
-- 插入一些默认数据
-- ================================
-- 插入默认语言列表
INSERT INTO `language_list` (`languageCode`, `languageName`, `nativeName`, `sortOrder`) VALUES
('auto', '自动检测', 'Auto Detect', 0),
('zh', '中文', '中文', 1),
('en', '英语', 'English', 2),
('ja', '日语', '日本語', 3),
('ko', '韩语', '한국어', 4),
('fr', '法语', 'Français', 5),
('de', '德语', 'Deutsch', 6),
('es', '西班牙语', 'Español', 7),
('ru', '俄语', 'Русский', 8),
('ar', '阿拉伯语', 'العربية', 9),
('pt', '葡萄牙语', 'Português', 10);
-- 插入默认翻译路由
INSERT INTO `translate_route` (`sourceLang`, `targetLang`, `translateService`, `priority`) VALUES
('auto', 'zh', 'google', 1),
('en', 'zh', 'google', 1),
('zh', 'en', 'google', 1),
('ja', 'zh', 'google', 1),
('ko', 'zh', 'google', 1);
-- 插入默认分组
INSERT INTO `group_manage` (`groupName`, `groupDescription`, `sortOrder`) VALUES
('常用回复', '常用的快捷回复', 1),
('问候语', '各种问候语', 2),
('业务回复', '业务相关回复', 3);
-- 插入一些默认参数
INSERT INTO `parameter` (`key`, `value`, `description`, `type`) VALUES
('system.version', '1.0.0', '系统版本', 'string'),
('translate.default_service', 'google', '默认翻译服务', 'string'),
('translate.cache_enabled', 'true', '是否启用翻译缓存', 'boolean'),
('proxy.global_enabled', 'false', '是否启用全局代理', 'boolean');
-- 恢复外键检查
SET FOREIGN_KEY_CHECKS = 1;
-- 显示创建的表
SHOW TABLES;

52
backend/create_missing_tables.sql Normal file
View File

@ -0,0 +1,52 @@
-- 创建缺失的表
-- 翻译路由表
CREATE TABLE IF NOT EXISTS `translate_route` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(100) NOT NULL UNIQUE,
`displayName` varchar(100) DEFAULT NULL,
`isEnabled` varchar(8) DEFAULT 'true',
`otherArgs` longtext DEFAULT NULL,
`created_at` datetime(6) NOT NULL,
`updated_at` datetime(6) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `name` (`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 语言列表表
CREATE TABLE IF NOT EXISTS `language_list` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`code` varchar(20) NOT NULL UNIQUE,
`name` varchar(100) NOT NULL,
`nativeName` varchar(100) DEFAULT NULL,
`isEnabled` varchar(8) DEFAULT 'true',
`created_at` datetime(6) NOT NULL,
`updated_at` datetime(6) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `code` (`code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 插入一些默认的翻译路由
INSERT IGNORE INTO `translate_route` (`name`, `displayName`, `isEnabled`, `otherArgs`, `created_at`, `updated_at`) VALUES
('youDao', '有道翻译', 'true', '{}', NOW(), NOW()),
('google', '谷歌翻译', 'true', '{}', NOW(), NOW()),
('deepl', 'DeepL翻译', 'true', '{}', NOW(), NOW()),
('baidu', '百度翻译', 'true', '{}', NOW(), NOW());
-- 插入一些默认的语言
INSERT IGNORE INTO `language_list` (`code`, `name`, `nativeName`, `isEnabled`, `created_at`, `updated_at`) VALUES
('auto', '自动检测', 'Auto Detect', 'true', NOW(), NOW()),
('zh-CN', '中文(简体)', '中文(简体)', 'true', NOW(), NOW()),
('zh-TW', '中文(繁体)', '中文(繁體)', 'true', NOW(), NOW()),
('en', '英语', 'English', 'true', NOW(), NOW()),
('ja', '日语', '日本語', 'true', NOW(), NOW()),
('ko', '韩语', '한국어', 'true', NOW(), NOW()),
('fr', '法语', 'Français', 'true', NOW(), NOW()),
('de', '德语', 'Deutsch', 'true', NOW(), NOW()),
('es', '西班牙语', 'Español', 'true', NOW(), NOW()),
('ru', '俄语', 'Русский', 'true', NOW(), NOW()),
('ar', '阿拉伯语', 'العربية', 'true', NOW(), NOW()),
('pt', '葡萄牙语', 'Português', 'true', NOW(), NOW()),
('it', '意大利语', 'Italiano', 'true', NOW(), NOW()),
('th', '泰语', 'ไทย', 'true', NOW(), NOW()),
('vi', '越南语', 'Tiếng Việt', 'true', NOW(), NOW());

93
backend/create_mysql_tables.py Normal file
View File

@ -0,0 +1,93 @@
#!/usr/bin/env python
"""
创建 MySQL 表结构脚本
"""
import pymysql
from pathlib import Path
def create_tables():
"""创建 MySQL 表结构"""
print("🔧 开始创建 MySQL 表结构...")
# MySQL 连接配置
mysql_config = {
'host': '192.168.1.114',
'port': 3306,
'user': 'seabox',
'password': '123456',
'database': 'fyapi',
'charset': 'utf8mb4'
}
try:
# 连接 MySQL
mysql_conn = pymysql.connect(**mysql_config)
mysql_cursor = mysql_conn.cursor()
print("✅ MySQL 连接成功")
# 读取 SQL 文件
sql_file_path = Path(__file__).parent / 'complete_mysql_import.sql'
if not sql_file_path.exists():
print(f"❌ SQL 文件不存在: {sql_file_path}")
return False
print(f"📍 读取 SQL 文件: {sql_file_path}")
with open(sql_file_path, 'r', encoding='utf-8') as f:
sql_content = f.read()
# 分割 SQL 语句并执行
sql_statements = [stmt.strip() for stmt in sql_content.split(';') if stmt.strip()]
executed_count = 0
for sql in sql_statements:
if sql.upper().startswith(('CREATE', 'DROP', 'INSERT', 'USE', 'SET', 'SHOW')):
try:
mysql_cursor.execute(sql)
executed_count += 1
if sql.upper().startswith('CREATE TABLE'):
table_name = sql.split('`')[1] if '`' in sql else 'unknown'
print(f"✅ 创建表: {table_name}")
elif sql.upper().startswith('INSERT'):
print(f"✅ 插入默认数据")
except Exception as e:
if "already exists" in str(e).lower():
print(f"⚠️ 表已存在，跳过")
else:
print(f"❌ 执行失败: {sql[:50]}... - {e}")
mysql_conn.commit()
print(f"\n✅ 表结构创建完成，执行了 {executed_count} 条 SQL 语句")
# 验证表是否创建成功
mysql_cursor.execute("SHOW TABLES")
tables = mysql_cursor.fetchall()
print(f"\n📋 数据库中的表 ({len(tables)} 个):")
for table in tables:
print(f" - {table[0]}")
mysql_conn.close()
return True
except Exception as e:
print(f"❌ 创建表结构失败: {e}")
return False
if __name__ == "__main__":
print("=" * 60)
print("MySQL 表结构创建工具")
print("=" * 60)
success = create_tables()
if success:
print("\n✅ 表结构创建成功！")
print("\n下一步:")
print("1. 运行数据迁移: python simple_data_migration.py")
print("2. 启动 Django 服务: python manage.py runserver")
else:
print("\n❌ 表结构创建失败，请检查错误信息")

92
backend/create_tables.py Normal file
View File

@ -0,0 +1,92 @@
#!/usr/bin/env python3
import os
import django
import pymysql
# 设置Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
django.setup()
from django.db import connection
def create_missing_tables():
"""创建缺失的表"""
sql_commands = [
# 翻译路由表
"""
CREATE TABLE IF NOT EXISTS `translate_route` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(100) NOT NULL,
`displayName` varchar(100) DEFAULT NULL,
`isEnabled` varchar(8) DEFAULT 'true',
`otherArgs` longtext DEFAULT NULL,
`created_at` datetime(6) NOT NULL,
`updated_at` datetime(6) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `name` (`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
""",
# 语言列表表
"""
CREATE TABLE IF NOT EXISTS `language_list` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`code` varchar(20) NOT NULL,
`name` varchar(100) NOT NULL,
`nativeName` varchar(100) DEFAULT NULL,
`isEnabled` varchar(8) DEFAULT 'true',
`created_at` datetime(6) NOT NULL,
`updated_at` datetime(6) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `code` (`code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
""",
# 插入默认翻译路由
"""
INSERT IGNORE INTO `translate_route` (`name`, `displayName`, `isEnabled`, `otherArgs`, `created_at`, `updated_at`) VALUES
('youDao', '有道翻译', 'true', '{}', NOW(), NOW()),
('google', '谷歌翻译', 'true', '{}', NOW(), NOW()),
('deepl', 'DeepL翻译', 'true', '{}', NOW(), NOW()),
('baidu', '百度翻译', 'true', '{}', NOW(), NOW());
""",
# 插入默认语言
"""
INSERT IGNORE INTO `language_list` (`code`, `name`, `nativeName`, `isEnabled`, `created_at`, `updated_at`) VALUES
('auto', '自动检测', 'Auto Detect', 'true', NOW(), NOW()),
('zh-CN', '中文(简体)', '中文(简体)', 'true', NOW(), NOW()),
('zh-TW', '中文(繁体)', '中文(繁體)', 'true', NOW(), NOW()),
('en', '英语', 'English', 'true', NOW(), NOW()),
('ja', '日语', '日本語', 'true', NOW(), NOW()),
('ko', '韩语', '한국어', 'true', NOW(), NOW()),
('fr', '法语', 'Français', 'true', NOW(), NOW()),
('de', '德语', 'Deutsch', 'true', NOW(), NOW()),
('es', '西班牙语', 'Español', 'true', NOW(), NOW()),
('ru', '俄语', 'Русский', 'true', NOW(), NOW()),
('ar', '阿拉伯语', 'العربية', 'true', NOW(), NOW()),
('pt', '葡萄牙语', 'Português', 'true', NOW(), NOW()),
('it', '意大利语', 'Italiano', 'true', NOW(), NOW()),
('th', '泰语', 'ไทย', 'true', NOW(), NOW()),
('vi', '越南语', 'Tiếng Việt', 'true', NOW(), NOW());
"""
]
try:
with connection.cursor() as cursor:
for sql in sql_commands:
if sql.strip():
print(f"执行SQL: {sql[:50]}...")
cursor.execute(sql)
print("✅ 执行成功")
print("🎉 所有表创建完成！")
return True
except Exception as e:
print(f"❌ 创建表失败: {e}")
return False
if __name__ == "__main__":
create_missing_tables()

74
backend/debug_sessions.py Normal file
View File

@ -0,0 +1,74 @@
#!/usr/bin/env python
"""
调试会话数据的脚本
"""
import os
import django
# 设置Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
django.setup()
from session.models import SessionList
from django.db import connection
def main():
print("🔍 调试会话数据存储问题")
print("=" * 50)
try:
# 1. 检查数据库连接
cursor = connection.cursor()
# 2. 检查表是否存在
cursor.execute("SHOW TABLES LIKE '%session%'")
tables = cursor.fetchall()
print(f"📋 会话相关表: {[t[0] for t in tables]}")
# 3. 检查session_list表结构
if any('session_list' in str(t) for t in tables):
cursor.execute("DESCRIBE session_list")
fields = cursor.fetchall()
print(f"\n📊 session_list 表结构:")
for field in fields:
print(f" {field[0]:<20} {field[1]:<20}")
# 4. 检查Django模型查询
print(f"\n🔍 Django模型查询:")
sessions = SessionList.objects.all()
print(f" 会话总数: {sessions.count()}")
if sessions.exists():
print(f" 前3个会话:")
for i, session in enumerate(sessions[:3], 1):
print(f" {i}. {session.platform} | {session.partitionId} | {session.createTime}")
else:
print(" ❌ 没有会话数据")
# 5. 直接SQL查询
print(f"\n🔍 直接SQL查询:")
cursor.execute("SELECT COUNT(*) FROM session_list")
count = cursor.fetchone()[0]
print(f" 直接查询记录数: {count}")
if count > 0:
cursor.execute("SELECT platform, partitionId, createTime FROM session_list LIMIT 3")
rows = cursor.fetchall()
print(f" 前3条记录:")
for i, row in enumerate(rows, 1):
print(f" {i}. {row[0]} | {row[1]} | {row[2]}")
# 6. 检查最近创建的记录
cursor.execute("SELECT platform, partitionId, created_at FROM session_list ORDER BY created_at DESC LIMIT 3")
recent_rows = cursor.fetchall()
print(f"\n📅 最近创建的3条记录:")
for i, row in enumerate(recent_rows, 1):
print(f" {i}. {row[0]} | {row[1]} | {row[2]}")
except Exception as e:
print(f"❌ 错误: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()

94
backend/fix_tables.py Normal file
View File

@ -0,0 +1,94 @@
#!/usr/bin/env python3
import os
import sys
import django
# 添加当前目录到Python路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
# 设置Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
django.setup()
from django.db import connection
def fix_tables():
"""修复表结构"""
sql_commands = [
# 删除旧表
"DROP TABLE IF EXISTS `translate_route`;",
"DROP TABLE IF EXISTS `language_list`;",
# 重新创建翻译路由表
"""
CREATE TABLE `translate_route` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(100) NOT NULL,
`displayName` varchar(100) DEFAULT NULL,
`isEnabled` varchar(8) DEFAULT 'true',
`otherArgs` longtext DEFAULT NULL,
`created_at` datetime(6) NOT NULL,
`updated_at` datetime(6) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `name` (`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
""",
# 重新创建语言列表表
"""
CREATE TABLE `language_list` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`code` varchar(20) NOT NULL,
`name` varchar(100) NOT NULL,
`nativeName` varchar(100) DEFAULT NULL,
`isEnabled` varchar(8) DEFAULT 'true',
`created_at` datetime(6) NOT NULL,
`updated_at` datetime(6) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `code` (`code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
""",
# 插入默认翻译路由
"""
INSERT INTO `translate_route` (`name`, `displayName`, `isEnabled`, `otherArgs`, `created_at`, `updated_at`) VALUES
('youDao', '有道翻译', 'true', '{}', NOW(), NOW()),
('google', '谷歌翻译', 'true', '{}', NOW(), NOW()),
('deepl', 'DeepL翻译', 'true', '{}', NOW(), NOW()),
('baidu', '百度翻译', 'true', '{}', NOW(), NOW());
""",
# 插入默认语言
"""
INSERT INTO `language_list` (`code`, `name`, `nativeName`, `isEnabled`, `created_at`, `updated_at`) VALUES
('auto', '自动检测', 'Auto Detect', 'true', NOW(), NOW()),
('zh-CN', '中文(简体)', '中文(简体)', 'true', NOW(), NOW()),
('zh-TW', '中文(繁体)', '中文(繁體)', 'true', NOW(), NOW()),
('en', '英语', 'English', 'true', NOW(), NOW()),
('ja', '日语', '日本語', 'true', NOW(), NOW()),
('ko', '韩语', '한국어', 'true', NOW(), NOW()),
('fr', '法语', 'Français', 'true', NOW(), NOW()),
('de', '德语', 'Deutsch', 'true', NOW(), NOW()),
('es', '西班牙语', 'Español', 'true', NOW(), NOW()),
('ru', '俄语', 'Русский', 'true', NOW(), NOW());
"""
]
try:
with connection.cursor() as cursor:
for sql in sql_commands:
if sql.strip():
print(f"执行SQL: {sql[:50]}...")
cursor.execute(sql)
print("✅ 执行成功")
print("🎉 表结构修复完成！")
return True
except Exception as e:
print(f"❌ 修复失败: {e}")
return False
if __name__ == "__main__":
fix_tables()

80
backend/fix_translate_config.py Normal file
View File

@ -0,0 +1,80 @@
#!/usr/bin/env python3
import os
import django
# 设置Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
django.setup()
from django.db import connection
def fix_translate_config_table():
"""修复 translate_config 表结构"""
sql_commands = [
# 删除旧的 translate_config 表
"DROP TABLE IF EXISTS `translate_config`;",
# 重新创建正确的 translate_config 表
"""
CREATE TABLE `translate_config` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`userId` varchar(255) DEFAULT NULL,
`platform` varchar(100) DEFAULT NULL,
`mode` varchar(20) DEFAULT 'cloud',
`translateRoute` varchar(50) DEFAULT NULL,
`receiveTranslateStatus` varchar(8) DEFAULT 'false',
`receiveSourceLanguage` varchar(20) DEFAULT 'auto',
`receiveTargetLanguage` varchar(20) DEFAULT 'zh-CN',
`sendTranslateStatus` varchar(8) DEFAULT 'true',
`sendSourceLanguage` varchar(20) DEFAULT 'auto',
`sendTargetLanguage` varchar(20) DEFAULT 'zh-CN',
`friendTranslateStatus` varchar(8) DEFAULT 'false',
`showAloneBtn` varchar(8) DEFAULT 'false',
`chineseDetectionStatus` varchar(8) DEFAULT 'false',
`translatePreview` varchar(8) DEFAULT 'false',
`interceptChinese` varchar(8) DEFAULT 'false',
`interceptLanguages` longtext DEFAULT NULL,
`translateHistory` varchar(8) DEFAULT 'false',
`autoTranslateGroupMessage` varchar(8) DEFAULT 'false',
`historyTranslateRoute` varchar(50) DEFAULT NULL,
`usePersonalConfig` varchar(8) DEFAULT 'true',
`created_at` datetime(6) NOT NULL,
`updated_at` datetime(6) NOT NULL,
PRIMARY KEY (`id`),
KEY `session_translateconfig_userId_platform_idx` (`userId`, `platform`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
""",
# 删除旧的 tg_sessions 表
"DROP TABLE IF EXISTS `tg_sessions`;",
# 重新创建正确的 tg_sessions 表
"""
CREATE TABLE `tg_sessions` (
`phoneNumber` varchar(50) NOT NULL,
`sessionStr` longtext DEFAULT NULL,
`created_at` datetime(6) NOT NULL,
`updated_at` datetime(6) NOT NULL,
PRIMARY KEY (`phoneNumber`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
"""
]
try:
with connection.cursor() as cursor:
for sql in sql_commands:
if sql.strip():
print(f"执行SQL: {sql[:50]}...")
cursor.execute(sql)
print("✅ 执行成功")
print("🎉 表结构修复完成！")
return True
except Exception as e:
print(f"❌ 修复失败: {e}")
return False
if __name__ == "__main__":
fix_translate_config_table()

323
backend/migrate_sqlite_data.py Normal file
View File

@ -0,0 +1,323 @@
#!/usr/bin/env python
"""
SQLite 数据迁移到 MySQL 脚本
"""
import os
import sys
import sqlite3
import django
import pymysql
from pathlib import Path
from datetime import datetime
# 设置Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
def create_mysql_tables():
"""创建 MySQL 表结构"""
print("🔧 创建 MySQL 表结构...")
# MySQL 连接配置
db_config = {
'host': '192.168.1.114',
'port': 3306,
'user': 'seabox',
'password': '123456',
'database': 'fyapi',
'charset': 'utf8mb4'
}
try:
connection = pymysql.connect(**db_config)
cursor = connection.cursor()
# 读取 SQL 文件并执行
sql_file_path = Path(__file__).parent / 'migrate_sqlite_to_mysql.sql'
if sql_file_path.exists():
with open(sql_file_path, 'r', encoding='utf-8') as f:
sql_content = f.read()
# 分割 SQL 语句并执行
sql_statements = [stmt.strip() for stmt in sql_content.split(';') if stmt.strip()]
for sql in sql_statements:
if sql.upper().startswith(('CREATE', 'USE', 'SELECT')):
try:
cursor.execute(sql)
print(f"✅ 执行成功: {sql[:50]}...")
except Exception as e:
if "already exists" in str(e).lower():
print(f"⚠️ 表已存在: {sql[:50]}...")
else:
print(f"❌ 执行失败: {sql[:50]}... - {e}")
connection.commit()
print("✅ MySQL 表结构创建完成")
else:
print("❌ SQL 文件不存在")
cursor.close()
connection.close()
except Exception as e:
print(f"❌ 创建 MySQL 表结构失败: {e}")
return False
return True
try:
# 先创建 MySQL 表结构
if not create_mysql_tables():
print("❌ 表结构创建失败，停止迁移")
sys.exit(1)
django.setup()
from session.models import (
SessionList, ContactInfo, FollowRecord, TranslateCache,
ProxyConfig, GlobalProxyConfig, GroupManage, QuickReplyRecord,
Parameter, TgSessions
)
from translate.models import TranslateConfig, TranslateRoute, LanguageList, UserQuickReplies
print("=" * 60)
print("SQLite 数据迁移到 MySQL")
print("=" * 60)
# SQLite 数据库路径
sqlite_db_path = Path(__file__).parent.parent.parent / 'liangzi_data' / 'session.db'
if not sqlite_db_path.exists():
print(f"❌ SQLite 数据库文件不存在: {sqlite_db_path}")
sys.exit(1)
print(f"📍 SQLite 数据库路径: {sqlite_db_path}")
# 连接 SQLite 数据库
sqlite_conn = sqlite3.connect(str(sqlite_db_path))
sqlite_conn.row_factory = sqlite3.Row # 使结果可以通过列名访问
cursor = sqlite_conn.cursor()
def migrate_table(sqlite_table, django_model, field_mapping=None, skip_duplicates=True):
"""迁移单个表的数据"""
try:
cursor.execute(f"SELECT * FROM {sqlite_table}")
rows = cursor.fetchall()
if not rows:
print(f"⚠️ 表 {sqlite_table} 没有数据")
return 0
migrated_count = 0
skipped_count = 0
for row in rows:
try:
# 转换数据
data = {}
for key in row.keys():
value = row[key]
# 应用字段映射
if field_mapping and key in field_mapping:
mapped_key = field_mapping[key]
if mapped_key: # 如果映射值不为 None
data[mapped_key] = value
else:
data[key] = value
# 移除 id 字段，让 Django 自动生成
if 'id' in data:
del data['id']
# 处理特殊字段
if sqlite_table == 'session_list' and 'partitionId' in data:
# 检查是否已存在相同的 partitionId
if skip_duplicates and django_model.objects.filter(partitionId=data['partitionId']).exists():
skipped_count += 1
continue
# 处理时间字段
if 'createTime' in data and data['createTime']:
# 保持原始时间格式
pass
# 处理布尔值字段
for field_name in ['windowStatus', 'onlineStatus', 'proxyStatus', 'userVerifyStatus', 'isTop']:
if field_name in data and data[field_name] is not None:
if isinstance(data[field_name], str):
data[field_name] = data[field_name].lower()
# 创建 Django 模型实例
instance = django_model(**data)
instance.save()
migrated_count += 1
except Exception as e:
print(f"❌ 迁移 {sqlite_table} 表的一行数据失败: {e}")
print(f" 数据: {data}")
continue
if skipped_count > 0:
print(f"✅ 表 {sqlite_table} 迁移完成，共 {migrated_count} 条记录，跳过 {skipped_count} 条重复记录")
else:
print(f"✅ 表 {sqlite_table} 迁移完成，共 {migrated_count} 条记录")
return migrated_count
except sqlite3.OperationalError as e:
if "no such table" in str(e):
print(f"⚠️ 表 {sqlite_table} 不存在，跳过")
return 0
else:
print(f"❌ 迁移表 {sqlite_table} 失败: {e}")
return 0
except Exception as e:
print(f"❌ 迁移表 {sqlite_table} 失败: {e}")
return 0
# 开始迁移
total_migrated = 0
migration_summary = {}
print("\n🚀 开始数据迁移...")
# 定义迁移顺序和配置
migration_tasks = [
('session_list', SessionList, None, "📋 会话列表"),
('translate_config', TranslateConfig, None, "🔧 翻译配置"),
('translate_route', TranslateRoute, None, "🛣️ 翻译路由"),
('language_list', LanguageList, None, "🌐 语言列表"),
('contact_info', ContactInfo, None, "👥 联系人信息"),
('follow_record', FollowRecord, None, "📝 跟进记录"),
('translate_cache', TranslateCache, None, "💾 翻译缓存"),
('proxy_config', ProxyConfig, None, "🔒 代理配置"),
('global_proxy_config', GlobalProxyConfig, None, "🌍 全局代理配置"),
('group_manage', GroupManage, None, "📁 分组管理"),
('quick_reply_record', QuickReplyRecord, None, "⚡ 快捷回复记录"),
('user_quick_replies', UserQuickReplies, None, "💬 用户快捷回复"),
('parameter', Parameter, {'key': 'key', 'value': 'value'}, "⚙️ 参数配置"),
('tg_sessions', TgSessions, None, "📱 Telegram 会话"),
]
# 执行迁移
for table_name, model_class, field_mapping, description in migration_tasks:
print(f"\n{description}...")
try:
migrated_count = migrate_table(table_name, model_class, field_mapping)
total_migrated += migrated_count
migration_summary[table_name] = {
'count': migrated_count,
'status': 'success',
'description': description
}
except Exception as e:
print(f"❌ 迁移 {table_name} 失败: {e}")
migration_summary[table_name] = {
'count': 0,
'status': 'failed',
'description': description,
'error': str(e)
}
# 关闭 SQLite 连接
sqlite_conn.close()
print("\n" + "=" * 80)
print(f"🎉 数据迁移完成！")
print(f"📊 总共迁移了 {total_migrated} 条记录")
print("=" * 80)
# 详细的迁移报告
print("\n📋 迁移详细报告:")
print("-" * 80)
success_count = 0
failed_count = 0
for table_name, info in migration_summary.items():
status_icon = "✅" if info['status'] == 'success' else "❌"
print(f"{status_icon} {info['description']}: {info['count']} 条记录")
if info['status'] == 'success':
success_count += 1
else:
failed_count += 1
if 'error' in info:
print(f" 错误: {info['error']}")
print("-" * 80)
print(f"成功迁移: {success_count} 个表")
print(f"失败迁移: {failed_count} 个表")
# 验证迁移结果
print("\n🔍 验证迁移结果...")
verification_results = {
'会话列表': SessionList.objects.count(),
'翻译配置': TranslateConfig.objects.count(),
'翻译路由': TranslateRoute.objects.count(),
'语言列表': LanguageList.objects.count(),
'联系人信息': ContactInfo.objects.count(),
'跟进记录': FollowRecord.objects.count(),
'翻译缓存': TranslateCache.objects.count(),
'代理配置': ProxyConfig.objects.count(),
'全局代理配置': GlobalProxyConfig.objects.count(),
'分组管理': GroupManage.objects.count(),
'快捷回复记录': QuickReplyRecord.objects.count(),
'用户快捷回复': UserQuickReplies.objects.count(),
'参数配置': Parameter.objects.count(),
'Telegram 会话': TgSessions.objects.count(),
}
print("-" * 50)
total_records = 0
for name, count in verification_results.items():
print(f"{name}: {count} 条")
total_records += count
print("-" * 50)
print(f"MySQL 数据库总记录数: {total_records} 条")
# 生成迁移报告文件
report_content = f"""
# SQLite 到 MySQL 数据迁移报告
## 迁移概要
- 迁移时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
- 总迁移记录数: {total_migrated}
- 成功迁移表数: {success_count}
- 失败迁移表数: {failed_count}
## 详细迁移结果
"""
for table_name, info in migration_summary.items():
status = "成功" if info['status'] == 'success' else "失败"
report_content += f"### {info['description']}\n"
report_content += f"- 表名: {table_name}\n"
report_content += f"- 状态: {status}\n"
report_content += f"- 迁移记录数: {info['count']}\n"
if 'error' in info:
report_content += f"- 错误信息: {info['error']}\n"
report_content += "\n"
report_content += "## MySQL 数据库验证结果\n\n"
for name, count in verification_results.items():
report_content += f"- {name}: {count} 条\n"
report_content += f"\n**总计: {total_records} 条记录**\n"
# 保存报告文件
report_file = Path(__file__).parent / f'migration_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.md'
with open(report_file, 'w', encoding='utf-8') as f:
f.write(report_content)
print(f"\n📄 迁移报告已保存到: {report_file}")
print("\n✅ 迁移验证完成！")
except Exception as e:
print(f"❌ 迁移过程中发生错误: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
print("开始执行迁移脚本...")

252
backend/migrate_sqlite_to_mysql.sql Normal file
View File

@ -0,0 +1,252 @@
-- ========================================
-- SQLite 到 MySQL 迁移脚本
-- ========================================
USE fyapi;
-- 1. 会话列表表
CREATE TABLE IF NOT EXISTS `session_list` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`partitionId` varchar(255) DEFAULT NULL,
`windowId` int(11) DEFAULT NULL,
`windowStatus` varchar(50) DEFAULT NULL,
`platform` varchar(100) DEFAULT NULL,
`onlineStatus` varchar(50) DEFAULT NULL,
`createTime` varchar(100) DEFAULT NULL,
`remarks` text,
`webUrl` text,
`avatarUrl` text,
`userName` varchar(255) DEFAULT NULL,
`nickName` varchar(255) DEFAULT NULL,
`msgCount` int(11) DEFAULT 0,
`userAgent` text,
`isTop` varchar(10) DEFAULT 'false',
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `partitionId` (`partitionId`),
KEY `platform` (`platform`),
KEY `windowStatus` (`windowStatus`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 2. 翻译配置表 (已存在，但需要确保字段完整)
CREATE TABLE IF NOT EXISTS `translate_config` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`userId` varchar(255) DEFAULT NULL,
`platform` varchar(100) DEFAULT NULL,
`mode` varchar(50) DEFAULT 'cloud',
`translateRoute` varchar(100) DEFAULT NULL,
`receiveTranslateStatus` varchar(10) DEFAULT 'false',
`receiveSourceLanguage` varchar(20) DEFAULT 'auto',
`receiveTargetLanguage` varchar(20) DEFAULT 'zh-CN',
`sendTranslateStatus` varchar(10) DEFAULT 'true',
`sendSourceLanguage` varchar(20) DEFAULT 'auto',
`sendTargetLanguage` varchar(20) DEFAULT 'en',
`friendTranslateStatus` varchar(10) DEFAULT 'true',
`showAloneBtn` varchar(10) DEFAULT 'true',
`chineseDetectionStatus` varchar(10) DEFAULT 'false',
`translatePreview` varchar(10) DEFAULT 'false',
`interceptChinese` varchar(10) DEFAULT 'false',
`interceptLanguages` text,
`translateHistory` varchar(10) DEFAULT 'false',
`autoTranslateGroupMessage` varchar(10) DEFAULT 'false',
`historyTranslateRoute` varchar(100) DEFAULT NULL,
`usePersonalConfig` varchar(10) DEFAULT 'true',
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `userId` (`userId`),
KEY `platform` (`platform`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 3. 翻译路由表 (已存在，但需要确保字段完整)
CREATE TABLE IF NOT EXISTS `translate_route` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(100) NOT NULL,
`zhName` varchar(255) DEFAULT NULL,
`enName` varchar(255) DEFAULT NULL,
`otherArgs` text,
`enable` tinyint(1) DEFAULT 1,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `name` (`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 4. 联系人信息表
CREATE TABLE IF NOT EXISTS `contact_info` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`userId` varchar(255) DEFAULT NULL,
`platform` varchar(100) DEFAULT NULL,
`phoneNumber` varchar(50) DEFAULT NULL,
`nickName` varchar(255) DEFAULT NULL,
`country` varchar(100) DEFAULT NULL,
`gender` varchar(20) DEFAULT NULL,
`gradeActivity` int(11) DEFAULT 0,
`customLevel` int(11) DEFAULT 0,
`remarks` text,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `userId` (`userId`),
KEY `platform` (`platform`),
KEY `phoneNumber` (`phoneNumber`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 5. 跟进记录表
CREATE TABLE IF NOT EXISTS `follow_record` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`userId` varchar(255) DEFAULT NULL,
`platform` varchar(100) DEFAULT NULL,
`content` text,
`timestamp` varchar(100) DEFAULT NULL,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `userId` (`userId`),
KEY `platform` (`platform`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 6. 语言列表表 (已存在，但需要确保字段完整)
CREATE TABLE IF NOT EXISTS `language_list` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`code` varchar(20) NOT NULL,
`zhName` varchar(255) DEFAULT NULL,
`enName` varchar(255) DEFAULT NULL,
`baidu` varchar(50) DEFAULT NULL,
`youDao` varchar(50) DEFAULT NULL,
`huoShan` varchar(50) DEFAULT NULL,
`xiaoNiu` varchar(50) DEFAULT NULL,
`google` varchar(50) DEFAULT NULL,
`tengXun` varchar(50) DEFAULT NULL,
`deepl` varchar(50) DEFAULT NULL,
`deepseek` varchar(50) DEFAULT NULL,
`bing` varchar(50) DEFAULT NULL,
`chatGpt4o` varchar(50) DEFAULT NULL,
`timestamp` varchar(100) DEFAULT NULL,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `code` (`code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 7. 翻译缓存表
CREATE TABLE IF NOT EXISTS `translate_cache` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`route` varchar(100) DEFAULT NULL,
`text` text,
`translateText` text,
`fromCode` varchar(20) DEFAULT NULL,
`toCode` varchar(20) DEFAULT NULL,
`partitionId` varchar(255) DEFAULT NULL,
`timestamp` varchar(100) DEFAULT NULL,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `route` (`route`),
KEY `partitionId` (`partitionId`),
KEY `fromCode_toCode` (`fromCode`, `toCode`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 8. 代理配置表
CREATE TABLE IF NOT EXISTS `proxy_config` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`partitionId` varchar(255) DEFAULT NULL,
`proxyStatus` varchar(10) DEFAULT 'false',
`proxyType` varchar(50) DEFAULT 'http',
`proxyIp` varchar(255) DEFAULT NULL,
`proxyPort` varchar(10) DEFAULT NULL,
`userVerifyStatus` varchar(10) DEFAULT 'false',
`username` varchar(255) DEFAULT NULL,
`password` varchar(255) DEFAULT NULL,
`timezone` varchar(100) DEFAULT NULL,
`defaultLanguage` varchar(20) DEFAULT NULL,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `partitionId` (`partitionId`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 9. 全局代理配置表
CREATE TABLE IF NOT EXISTS `global_proxy_config` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`proxyStatus` varchar(10) DEFAULT 'false',
`proxyType` varchar(50) DEFAULT 'http',
`proxyIp` varchar(255) DEFAULT NULL,
`proxyPort` varchar(10) DEFAULT NULL,
`userVerifyStatus` varchar(10) DEFAULT 'false',
`username` varchar(255) DEFAULT NULL,
`password` varchar(255) DEFAULT NULL,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 10. 分组管理表
CREATE TABLE IF NOT EXISTS `group_manage` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(255) NOT NULL,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 11. 快捷回复记录表
CREATE TABLE IF NOT EXISTS `quick_reply_record` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`groupId` varchar(255) DEFAULT NULL,
`type` varchar(50) DEFAULT NULL,
`remark` varchar(255) DEFAULT NULL,
`content` text,
`url` text,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `groupId` (`groupId`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 12. 用户快捷回复表 (已存在，但需要确保字段完整)
CREATE TABLE IF NOT EXISTS `user_quick_replies` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`partitionId` varchar(255) DEFAULT NULL,
`userId` varchar(255) DEFAULT NULL,
`content` text NOT NULL,
`remark` varchar(255) DEFAULT NULL,
`sendMode` varchar(50) DEFAULT 'direct',
`sortOrder` int(11) DEFAULT 0,
`isEnabled` tinyint(1) DEFAULT 1,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `partitionId` (`partitionId`),
KEY `userId` (`userId`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 13. 参数配置表
CREATE TABLE IF NOT EXISTS `parameter` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`key` varchar(255) NOT NULL,
`value` text,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `key` (`key`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 14. Telegram 会话表
CREATE TABLE IF NOT EXISTS `tg_sessions` (
`phoneNumber` varchar(50) NOT NULL,
`sessionStr` text,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
`updated_at` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`phoneNumber`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 添加索引以提高查询性能
CREATE INDEX IF NOT EXISTS `idx_session_list_platform_status` ON `session_list` (`platform`, `windowStatus`);
CREATE INDEX IF NOT EXISTS `idx_translate_config_user_platform` ON `translate_config` (`userId`, `platform`);
CREATE INDEX IF NOT EXISTS `idx_contact_info_user_platform` ON `contact_info` (`userId`, `platform`);
CREATE INDEX IF NOT EXISTS `idx_translate_cache_partition` ON `translate_cache` (`partitionId`);
-- 完成提示
SELECT 'MySQL 表结构创建完成！' as status;

1
backend/session/__init__.py Normal file
View File

@ -0,0 +1 @@
# Session 应用初始化文件

7
backend/session/apps.py Normal file
View File

@ -0,0 +1,7 @@
from django.apps import AppConfig
class SessionConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'session'
verbose_name = '会话管理'

1
backend/session/management/__init__.py Normal file
View File

@ -0,0 +1 @@
# Django management commands

1
backend/session/management/commands/__init__.py Normal file
View File

@ -0,0 +1 @@
# Django management commands

229
backend/session/management/commands/migrate_sqlite_to_mysql.py Normal file
View File

@ -0,0 +1,229 @@
from django.core.management.base import BaseCommand
import sqlite3
import pymysql
from pathlib import Path
from datetime import datetime
from session.models import (
SessionList, ContactInfo, FollowRecord, TranslateCache,
ProxyConfig, GlobalProxyConfig, GroupManage, QuickReplyRecord,
Parameter, TgSessions
)
from translate.models import TranslateConfig, TranslateRoute, LanguageList, UserQuickReplies
class Command(BaseCommand):
help = 'Migrate data from SQLite to MySQL'
def add_arguments(self, parser):
parser.add_argument(
'--sqlite-path',
type=str,
help='Path to SQLite database file',
default=None
)
parser.add_argument(
'--dry-run',
action='store_true',
help='Run migration in dry-run mode (no actual data changes)',
)
def handle(self, *args, **options):
self.stdout.write(self.style.SUCCESS('🚀 开始 SQLite 到 MySQL 数据迁移...'))
# 确定SQLite数据库路径
if options['sqlite_path']:
sqlite_db_path = Path(options['sqlite_path'])
else:
sqlite_db_path = Path(__file__).parent.parent.parent.parent.parent.parent / 'liangzi_data' / 'session.db'
self.stdout.write(f"📍 SQLite 数据库路径: {sqlite_db_path}")
if not sqlite_db_path.exists():
self.stdout.write(self.style.ERROR(f"❌ SQLite 数据库文件不存在: {sqlite_db_path}"))
return
# 连接SQLite数据库
try:
sqlite_conn = sqlite3.connect(str(sqlite_db_path))
sqlite_conn.row_factory = sqlite3.Row
cursor = sqlite_conn.cursor()
self.stdout.write(self.style.SUCCESS("✅ SQLite 数据库连接成功"))
except Exception as e:
self.stdout.write(self.style.ERROR(f"❌ SQLite 数据库连接失败: {e}"))
return
# 检查SQLite表
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = cursor.fetchall()
self.stdout.write(f"📋 SQLite 数据库中的表: {[table[0] for table in tables]}")
# 定义迁移任务
migration_tasks = [
('session_list', SessionList, None, "📋 会话列表"),
('translate_config', TranslateConfig, None, "🔧 翻译配置"),
('translate_route', TranslateRoute, None, "🛣️ 翻译路由"),
('language_list', LanguageList, None, "🌐 语言列表"),
('contact_info', ContactInfo, None, "👥 联系人信息"),
('follow_record', FollowRecord, None, "📝 跟进记录"),
('translate_cache', TranslateCache, None, "💾 翻译缓存"),
('proxy_config', ProxyConfig, None, "🔒 代理配置"),
('global_proxy_config', GlobalProxyConfig, None, "🌍 全局代理配置"),
('group_manage', GroupManage, None, "📁 分组管理"),
('quick_reply_record', QuickReplyRecord, None, "⚡ 快捷回复记录"),
('user_quick_replies', UserQuickReplies, None, "💬 用户快捷回复"),
('parameter', Parameter, {'key': 'key', 'value': 'value'}, "⚙️ 参数配置"),
('tg_sessions', TgSessions, None, "📱 Telegram 会话"),
]
total_migrated = 0
migration_summary = {}
# 执行迁移
for table_name, model_class, field_mapping, description in migration_tasks:
self.stdout.write(f"\n{description}...")
try:
migrated_count = self.migrate_table(cursor, table_name, model_class, field_mapping, options['dry_run'])
total_migrated += migrated_count
migration_summary[table_name] = {
'count': migrated_count,
'status': 'success',
'description': description
}
except Exception as e:
self.stdout.write(self.style.ERROR(f"❌ 迁移 {table_name} 失败: {e}"))
migration_summary[table_name] = {
'count': 0,
'status': 'failed',
'description': description,
'error': str(e)
}
# 关闭SQLite连接
sqlite_conn.close()
# 输出迁移报告
self.stdout.write("\n" + "=" * 80)
self.stdout.write(self.style.SUCCESS(f"🎉 数据迁移完成！"))
self.stdout.write(f"📊 总共迁移了 {total_migrated} 条记录")
self.stdout.write("=" * 80)
# 详细报告
self.stdout.write("\n📋 迁移详细报告:")
self.stdout.write("-" * 80)
success_count = 0
failed_count = 0
for table_name, info in migration_summary.items():
if info['status'] == 'success':
self.stdout.write(self.style.SUCCESS(f"✅ {info['description']}: {info['count']} 条记录"))
success_count += 1
else:
self.stdout.write(self.style.ERROR(f"❌ {info['description']}: {info['count']} 条记录"))
failed_count += 1
if 'error' in info:
self.stdout.write(f" 错误: {info['error']}")
self.stdout.write("-" * 80)
self.stdout.write(f"成功迁移: {success_count} 个表")
self.stdout.write(f"失败迁移: {failed_count} 个表")
# 验证迁移结果
if not options['dry_run']:
self.stdout.write("\n🔍 验证迁移结果...")
verification_results = {
'会话列表': SessionList.objects.count(),
'翻译配置': TranslateConfig.objects.count(),
'翻译路由': TranslateRoute.objects.count(),
'语言列表': LanguageList.objects.count(),
'联系人信息': ContactInfo.objects.count(),
'跟进记录': FollowRecord.objects.count(),
'翻译缓存': TranslateCache.objects.count(),
'代理配置': ProxyConfig.objects.count(),
'全局代理配置': GlobalProxyConfig.objects.count(),
'分组管理': GroupManage.objects.count(),
'快捷回复记录': QuickReplyRecord.objects.count(),
'用户快捷回复': UserQuickReplies.objects.count(),
'参数配置': Parameter.objects.count(),
'Telegram 会话': TgSessions.objects.count(),
}
self.stdout.write("-" * 50)
total_records = 0
for name, count in verification_results.items():
self.stdout.write(f"{name}: {count} 条")
total_records += count
self.stdout.write("-" * 50)
self.stdout.write(f"MySQL 数据库总记录数: {total_records} 条")
self.stdout.write(self.style.SUCCESS("\n✅ 迁移完成！"))
def migrate_table(self, cursor, sqlite_table, django_model, field_mapping=None, dry_run=False):
"""迁移单个表的数据"""
try:
cursor.execute(f"SELECT * FROM {sqlite_table}")
rows = cursor.fetchall()
if not rows:
self.stdout.write(f"⚠️ 表 {sqlite_table} 没有数据")
return 0
migrated_count = 0
skipped_count = 0
for row in rows:
try:
# 转换数据
data = {}
for key in row.keys():
value = row[key]
# 应用字段映射
if field_mapping and key in field_mapping:
mapped_key = field_mapping[key]
if mapped_key:
data[mapped_key] = value
else:
data[key] = value
# 移除id字段，让Django自动生成
if 'id' in data:
del data['id']
# 处理特殊字段
if sqlite_table == 'session_list' and 'partitionId' in data:
# 检查是否已存在相同的partitionId
if django_model.objects.filter(partitionId=data['partitionId']).exists():
skipped_count += 1
continue
# 处理布尔值字段
for field_name in ['windowStatus', 'onlineStatus', 'proxyStatus', 'userVerifyStatus', 'isTop']:
if field_name in data and data[field_name] is not None:
if isinstance(data[field_name], str):
data[field_name] = data[field_name].lower()
if not dry_run:
# 创建Django模型实例
instance = django_model(**data)
instance.save()
migrated_count += 1
except Exception as e:
self.stdout.write(self.style.ERROR(f"❌ 迁移 {sqlite_table} 表的一行数据失败: {e}"))
continue
if skipped_count > 0:
self.stdout.write(f"✅ 表 {sqlite_table} 迁移完成，共 {migrated_count} 条记录，跳过 {skipped_count} 条重复记录")
else:
self.stdout.write(f"✅ 表 {sqlite_table} 迁移完成，共 {migrated_count} 条记录")
return migrated_count
except sqlite3.OperationalError as e:
if "no such table" in str(e):
self.stdout.write(f"⚠️ 表 {sqlite_table} 不存在，跳过")
return 0
else:
raise e
except Exception as e:
raise e

0
backend/session/migrations/__init__.py Normal file
View File

298
backend/session/models.py Normal file
View File

@ -0,0 +1,298 @@
from django.db import models
from django.utils import timezone
class SessionList(models.Model):
"""会话列表模型"""
partitionId = models.CharField(max_length=255, unique=True, null=True, blank=True, verbose_name="分区ID")
windowId = models.IntegerField(null=True, blank=True, verbose_name="窗口ID")
windowStatus = models.CharField(max_length=50, null=True, blank=True, verbose_name="窗口状态")
platform = models.CharField(max_length=100, null=True, blank=True, verbose_name="平台")
onlineStatus = models.CharField(max_length=50, null=True, blank=True, verbose_name="在线状态")
createTime = models.CharField(max_length=100, null=True, blank=True, verbose_name="创建时间")
remarks = models.TextField(null=True, blank=True, verbose_name="备注")
webUrl = models.TextField(null=True, blank=True, verbose_name="网址")
avatarUrl = models.TextField(null=True, blank=True, verbose_name="头像URL")
userName = models.CharField(max_length=255, null=True, blank=True, verbose_name="用户名")
nickName = models.CharField(max_length=255, null=True, blank=True, verbose_name="昵称")
msgCount = models.IntegerField(default=0, verbose_name="消息数量")
userAgent = models.TextField(null=True, blank=True, verbose_name="用户代理")
isTop = models.CharField(max_length=10, default='false', verbose_name="是否置顶")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'session_list'
verbose_name = '会话列表'
verbose_name_plural = '会话列表'
indexes = [
models.Index(fields=['platform', 'windowStatus']),
models.Index(fields=['partitionId']),
]
def __str__(self):
return f"{self.platform} - {self.nickName or self.userName}"
class ContactInfo(models.Model):
"""联系人信息模型"""
userId = models.CharField(max_length=255, null=True, blank=True, verbose_name="用户ID")
platform = models.CharField(max_length=100, null=True, blank=True, verbose_name="平台")
phoneNumber = models.CharField(max_length=50, null=True, blank=True, verbose_name="电话号码")
nickName = models.CharField(max_length=255, null=True, blank=True, verbose_name="昵称")
country = models.CharField(max_length=100, null=True, blank=True, verbose_name="国家")
gender = models.CharField(max_length=20, null=True, blank=True, verbose_name="性别")
gradeActivity = models.IntegerField(default=0, verbose_name="等级活动")
customLevel = models.IntegerField(default=0, verbose_name="自定义等级")
remarks = models.TextField(null=True, blank=True, verbose_name="备注")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'contact_info'
verbose_name = '联系人信息'
verbose_name_plural = '联系人信息'
indexes = [
models.Index(fields=['userId', 'platform']),
models.Index(fields=['phoneNumber']),
]
def __str__(self):
return f"{self.nickName or self.userId} - {self.platform}"
class FollowRecord(models.Model):
"""跟进记录模型"""
userId = models.CharField(max_length=255, null=True, blank=True, verbose_name="用户ID")
platform = models.CharField(max_length=100, null=True, blank=True, verbose_name="平台")
content = models.TextField(null=True, blank=True, verbose_name="内容")
timestamp = models.CharField(max_length=100, null=True, blank=True, verbose_name="时间戳")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'follow_record'
verbose_name = '跟进记录'
verbose_name_plural = '跟进记录'
indexes = [
models.Index(fields=['userId', 'platform']),
]
def __str__(self):
return f"{self.userId} - {self.platform}"
class TranslateCache(models.Model):
"""翻译缓存模型"""
route = models.CharField(max_length=100, null=True, blank=True, verbose_name="路由")
text = models.TextField(null=True, blank=True, verbose_name="原文")
translateText = models.TextField(null=True, blank=True, verbose_name="译文")
fromCode = models.CharField(max_length=20, null=True, blank=True, verbose_name="源语言")
toCode = models.CharField(max_length=20, null=True, blank=True, verbose_name="目标语言")
partitionId = models.CharField(max_length=255, null=True, blank=True, verbose_name="分区ID")
timestamp = models.CharField(max_length=100, null=True, blank=True, verbose_name="时间戳")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'translate_cache'
verbose_name = '翻译缓存'
verbose_name_plural = '翻译缓存'
indexes = [
models.Index(fields=['route']),
models.Index(fields=['partitionId']),
models.Index(fields=['fromCode', 'toCode']),
]
def __str__(self):
return f"{self.route} - {self.fromCode} to {self.toCode}"
class ProxyConfig(models.Model):
"""代理配置模型"""
partitionId = models.CharField(max_length=255, null=True, blank=True, verbose_name="分区ID")
proxyStatus = models.CharField(max_length=10, default='false', verbose_name="代理状态")
proxyType = models.CharField(max_length=50, default='http', verbose_name="代理类型")
proxyIp = models.CharField(max_length=255, null=True, blank=True, verbose_name="代理IP")
proxyPort = models.CharField(max_length=10, null=True, blank=True, verbose_name="代理端口")
userVerifyStatus = models.CharField(max_length=10, default='false', verbose_name="用户验证状态")
username = models.CharField(max_length=255, null=True, blank=True, verbose_name="用户名")
password = models.CharField(max_length=255, null=True, blank=True, verbose_name="密码")
timezone = models.CharField(max_length=100, null=True, blank=True, verbose_name="时区")
defaultLanguage = models.CharField(max_length=20, null=True, blank=True, verbose_name="默认语言")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'proxy_config'
verbose_name = '代理配置'
verbose_name_plural = '代理配置'
indexes = [
models.Index(fields=['partitionId']),
]
def __str__(self):
return f"{self.partitionId} - {self.proxyType}"
class GlobalProxyConfig(models.Model):
"""全局代理配置模型"""
proxyStatus = models.CharField(max_length=10, default='false', verbose_name="代理状态")
proxyType = models.CharField(max_length=50, default='http', verbose_name="代理类型")
proxyIp = models.CharField(max_length=255, null=True, blank=True, verbose_name="代理IP")
proxyPort = models.CharField(max_length=10, null=True, blank=True, verbose_name="代理端口")
userVerifyStatus = models.CharField(max_length=10, default='false', verbose_name="用户验证状态")
username = models.CharField(max_length=255, null=True, blank=True, verbose_name="用户名")
password = models.CharField(max_length=255, null=True, blank=True, verbose_name="密码")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'global_proxy_config'
verbose_name = '全局代理配置'
verbose_name_plural = '全局代理配置'
def __str__(self):
return f"全局代理 - {self.proxyType}"
class GroupManage(models.Model):
"""分组管理模型"""
name = models.CharField(max_length=255, verbose_name="分组名称")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'group_manage'
verbose_name = '分组管理'
verbose_name_plural = '分组管理'
def __str__(self):
return self.name
class QuickReplyRecord(models.Model):
"""快捷回复记录模型"""
groupId = models.CharField(max_length=255, null=True, blank=True, verbose_name="分组ID")
type = models.CharField(max_length=50, null=True, blank=True, verbose_name="类型")
remark = models.CharField(max_length=255, null=True, blank=True, verbose_name="备注")
content = models.TextField(null=True, blank=True, verbose_name="内容")
url = models.TextField(null=True, blank=True, verbose_name="URL")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'quick_reply_record'
verbose_name = '快捷回复记录'
verbose_name_plural = '快捷回复记录'
indexes = [
models.Index(fields=['groupId']),
]
def __str__(self):
return f"{self.remark or self.content[:50]}"
class Parameter(models.Model):
"""参数配置模型"""
key = models.CharField(max_length=255, unique=True, verbose_name="参数键")
value = models.TextField(null=True, blank=True, verbose_name="参数值")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'parameter'
verbose_name = '参数配置'
verbose_name_plural = '参数配置'
def __str__(self):
return f"{self.key} = {self.value}"
class TgSessions(models.Model):
"""Telegram 会话模型"""
phoneNumber = models.CharField(max_length=50, primary_key=True, verbose_name="电话号码")
sessionStr = models.TextField(null=True, blank=True, verbose_name="会话字符串")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'tg_sessions'
verbose_name = 'Telegram 会话'
verbose_name_plural = 'Telegram 会话'
def __str__(self):
return self.phoneNumber
class TranslateConfig(models.Model):
"""翻译配置模型"""
userId = models.CharField(max_length=255, null=True, blank=True, verbose_name="用户ID")
platform = models.CharField(max_length=100, null=True, blank=True, verbose_name="平台")
mode = models.CharField(max_length=20, default='cloud', verbose_name="翻译模式")
translateRoute = models.CharField(max_length=50, null=True, blank=True, verbose_name="翻译路由")
receiveTranslateStatus = models.CharField(max_length=8, default='false', verbose_name="接收翻译状态")
receiveSourceLanguage = models.CharField(max_length=20, default='auto', verbose_name="接收源语言")
receiveTargetLanguage = models.CharField(max_length=20, default='zh-CN', verbose_name="接收目标语言")
sendTranslateStatus = models.CharField(max_length=8, default='true', verbose_name="发送翻译状态")
sendSourceLanguage = models.CharField(max_length=20, default='auto', verbose_name="发送源语言")
sendTargetLanguage = models.CharField(max_length=20, default='zh-CN', verbose_name="发送目标语言")
friendTranslateStatus = models.CharField(max_length=8, default='false', verbose_name="好友翻译状态")
showAloneBtn = models.CharField(max_length=8, default='false', verbose_name="显示独立按钮")
chineseDetectionStatus = models.CharField(max_length=8, default='false', verbose_name="中文检测状态")
translatePreview = models.CharField(max_length=8, default='false', verbose_name="翻译预览")
interceptChinese = models.CharField(max_length=8, default='false', verbose_name="拦截中文")
interceptLanguages = models.TextField(null=True, blank=True, verbose_name="拦截语言")
translateHistory = models.CharField(max_length=8, default='false', verbose_name="翻译历史")
autoTranslateGroupMessage = models.CharField(max_length=8, default='false', verbose_name="自动翻译群消息")
historyTranslateRoute = models.CharField(max_length=50, null=True, blank=True, verbose_name="历史翻译路由")
usePersonalConfig = models.CharField(max_length=8, default='true', verbose_name="使用个人配置")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'translate_config'
verbose_name = '翻译配置'
verbose_name_plural = '翻译配置'
indexes = [
models.Index(fields=['userId', 'platform']),
]
def __str__(self):
return f"{self.userId or self.platform} - {self.translateRoute}"
class TranslateRoute(models.Model):
"""翻译路由配置模型"""
name = models.CharField(max_length=100, unique=True, verbose_name="路由名称")
displayName = models.CharField(max_length=100, null=True, blank=True, verbose_name="显示名称")
isEnabled = models.CharField(max_length=8, default='true', verbose_name="是否启用")
otherArgs = models.TextField(null=True, blank=True, verbose_name="其他参数")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'translate_route'
verbose_name = '翻译路由'
verbose_name_plural = '翻译路由'
def __str__(self):
return f"{self.displayName or self.name}"
class LanguageList(models.Model):
"""语言列表模型"""
code = models.CharField(max_length=20, unique=True, verbose_name="语言代码")
name = models.CharField(max_length=100, verbose_name="语言名称")
nativeName = models.CharField(max_length=100, null=True, blank=True, verbose_name="本地名称")
isEnabled = models.CharField(max_length=8, default='true', verbose_name="是否启用")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
class Meta:
db_table = 'language_list'
verbose_name = '语言列表'
verbose_name_plural = '语言列表'
def __str__(self):
return f"{self.name} ({self.code})"

131
backend/session/serializers.py Normal file
View File

@ -0,0 +1,131 @@
from rest_framework import serializers
from .models import (
SessionList, ContactInfo, FollowRecord, TranslateCache,
ProxyConfig, GlobalProxyConfig, GroupManage, QuickReplyRecord,
Parameter, TgSessions, TranslateConfig, TranslateRoute, LanguageList
)
class SessionListSerializer(serializers.ModelSerializer):
"""会话列表序列化器"""
class Meta:
model = SessionList
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class ContactInfoSerializer(serializers.ModelSerializer):
"""联系人信息序列化器"""
class Meta:
model = ContactInfo
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class FollowRecordSerializer(serializers.ModelSerializer):
"""跟进记录序列化器"""
class Meta:
model = FollowRecord
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class TranslateCacheSerializer(serializers.ModelSerializer):
"""翻译缓存序列化器"""
class Meta:
model = TranslateCache
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class ProxyConfigSerializer(serializers.ModelSerializer):
"""代理配置序列化器"""
class Meta:
model = ProxyConfig
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class GlobalProxyConfigSerializer(serializers.ModelSerializer):
"""全局代理配置序列化器"""
class Meta:
model = GlobalProxyConfig
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class GroupManageSerializer(serializers.ModelSerializer):
"""分组管理序列化器"""
class Meta:
model = GroupManage
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class QuickReplyRecordSerializer(serializers.ModelSerializer):
"""快捷回复记录序列化器"""
class Meta:
model = QuickReplyRecord
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class ParameterSerializer(serializers.ModelSerializer):
"""参数配置序列化器"""
class Meta:
model = Parameter
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class TgSessionsSerializer(serializers.ModelSerializer):
"""Telegram 会话序列化器"""
class Meta:
model = TgSessions
fields = '__all__'
read_only_fields = ('created_at', 'updated_at')
# 用于生成唯一分区ID的序列化器
class GeneratePartitionIdSerializer(serializers.Serializer):
"""生成分区ID序列化器"""
platform = serializers.CharField(max_length=100, required=True, help_text="平台名称")
url = serializers.URLField(required=False, allow_blank=True, help_text="自定义网址")
nickname = serializers.CharField(max_length=255, required=False, allow_blank=True, help_text="昵称")
class TranslateConfigSerializer(serializers.ModelSerializer):
"""翻译配置序列化器"""
class Meta:
model = TranslateConfig
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class TranslateRouteSerializer(serializers.ModelSerializer):
"""翻译路由序列化器"""
class Meta:
model = TranslateRoute
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')
class LanguageListSerializer(serializers.ModelSerializer):
"""语言列表序列化器"""
class Meta:
model = LanguageList
fields = '__all__'
read_only_fields = ('id', 'created_at', 'updated_at')

28
backend/session/urls.py Normal file
View File

@ -0,0 +1,28 @@
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import (
SessionListViewSet, ContactInfoViewSet, FollowRecordViewSet,
TranslateCacheViewSet, ProxyConfigViewSet, GlobalProxyConfigViewSet,
GroupManageViewSet, QuickReplyRecordViewSet, ParameterViewSet,
TgSessionsViewSet, TranslateConfigViewSet, TranslateRouteViewSet,
LanguageListViewSet
)
router = DefaultRouter()
router.register(r'sessions', SessionListViewSet, basename='sessions')
router.register(r'contacts', ContactInfoViewSet, basename='contacts')
router.register(r'follow-records', FollowRecordViewSet, basename='follow-records')
router.register(r'translate-cache', TranslateCacheViewSet, basename='translate-cache')
router.register(r'proxy-config', ProxyConfigViewSet, basename='proxy-config')
router.register(r'global-proxy-config', GlobalProxyConfigViewSet, basename='global-proxy-config')
router.register(r'groups', GroupManageViewSet, basename='groups')
router.register(r'quick-replies', QuickReplyRecordViewSet, basename='quick-replies')
router.register(r'parameters', ParameterViewSet, basename='parameters')
router.register(r'tg-sessions', TgSessionsViewSet, basename='tg-sessions')
router.register(r'translate-config', TranslateConfigViewSet, basename='translate-config')
router.register(r'translate-route', TranslateRouteViewSet, basename='translate-route')
router.register(r'language-list', LanguageListViewSet, basename='language-list')
urlpatterns = [
path('api/', include(router.urls)),
]

469
backend/session/views.py Normal file
View File

@ -0,0 +1,469 @@
import random
import string
from datetime import datetime
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import AllowAny
from django.db import transaction
from django.shortcuts import get_object_or_404
from .models import (
SessionList, ContactInfo, FollowRecord, TranslateCache,
ProxyConfig, GlobalProxyConfig, GroupManage, QuickReplyRecord,
Parameter, TgSessions, TranslateConfig, TranslateRoute, LanguageList
)
from .serializers import (
SessionListSerializer, ContactInfoSerializer, FollowRecordSerializer,
TranslateCacheSerializer, ProxyConfigSerializer, GlobalProxyConfigSerializer,
GroupManageSerializer, QuickReplyRecordSerializer, ParameterSerializer,
TgSessionsSerializer, GeneratePartitionIdSerializer, TranslateConfigSerializer,
TranslateRouteSerializer, LanguageListSerializer
)
def generate_unique_partition_id(length=8, max_retry=10):
"""生成唯一的分区ID"""
chars = string.ascii_letters + string.digits
for _ in range(max_retry):
partition_id = ''.join(random.choices(chars, k=length))
if not SessionList.objects.filter(partitionId=partition_id).exists():
return partition_id
raise Exception(f"Failed to generate unique partition ID after {max_retry} attempts")
class SessionListViewSet(viewsets.ModelViewSet):
"""会话列表视图集"""
queryset = SessionList.objects.all()
serializer_class = SessionListSerializer
permission_classes = [AllowAny]
@action(detail=False, methods=['get'], url_path='get_by_partition_id')
def get_by_partition_id(self, request):
"""根据分区ID获取会话信息"""
partition_id = request.query_params.get('partitionId')
platform = request.query_params.get('platform')
if not partition_id:
return Response({
'status': False,
'message': '缺少partitionId参数'
}, status=status.HTTP_400_BAD_REQUEST)
try:
# 根据partitionId查找会话，如果提供了platform也加入查询条件
filter_kwargs = {'partitionId': partition_id}
if platform:
filter_kwargs['platform'] = platform
session = SessionList.objects.filter(**filter_kwargs).first()
if not session:
return Response({
'status': False,
'message': '没有找到对应的会话记录'
}, status=status.HTTP_404_NOT_FOUND)
serializer = self.get_serializer(session)
return Response({
'status': True,
'message': 'success',
'data': {
'session': serializer.data
}
})
except Exception as e:
return Response({
'status': False,
'message': f'查询失败：{str(e)}'
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
def get_queryset(self):
"""根据平台过滤会话列表"""
queryset = SessionList.objects.all()
platform = self.request.query_params.get('platform', None)
if platform:
queryset = queryset.filter(platform=platform)
# 使用id排序，避免created_at字段问题
return queryset.order_by('-id')
@action(detail=False, methods=['post'])
@transaction.atomic
def add_session(self, request):
"""添加新会话"""
try:
platform = request.data.get('platform')
url = request.data.get('url', '')
nickname = request.data.get('nickname', '')
if not platform:
return Response({
'status': False,
'message': '平台参数不能为空'
}, status=status.HTTP_400_BAD_REQUEST)
# 生成唯一分区ID
partition_id = generate_unique_partition_id()
create_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 创建会话数据
session_data = {
'partitionId': partition_id,
'windowStatus': 'false',
'createTime': create_time,
'platform': platform,
'msgCount': 0,
'onlineStatus': 'false',
'windowId': 0
}
if platform == 'CustomWeb':
session_data.update({
'nickName': nickname,
'webUrl': url
})
else:
# 预定义的平台URL
platform_urls = {
'Telegram': 'https://web.telegram.org/a/',
'WhatsApp': 'https://web.whatsapp.com/',
'TikTok': 'https://www.tiktok.com/messages?lang=zh-Hans'
}
session_data.update({
'nickName': platform.lower(),
'webUrl': platform_urls.get(platform, '')
})
# 保存到数据库
serializer = self.get_serializer(data=session_data)
if serializer.is_valid():
session = serializer.save()
return Response({
'status': True,
'message': '新增成功',
'data': {'partitionId': partition_id}
})
else:
return Response({
'status': False,
'message': f'数据验证失败: {serializer.errors}'
}, status=status.HTTP_400_BAD_REQUEST)
except Exception as e:
return Response({
'status': False,
'message': f'添加失败：{str(e)}'
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@action(detail=False, methods=['get'])
def get_by_partition_id(self, request):
"""根据分区ID获取会话信息"""
partition_id = request.query_params.get('partitionId')
platform = request.query_params.get('platform')
if not partition_id:
return Response({
'status': False,
'message': '分区ID不能为空'
}, status=status.HTTP_400_BAD_REQUEST)
try:
session = SessionList.objects.get(partitionId=partition_id)
serializer = self.get_serializer(session)
return Response({
'status': True,
'message': '查询成功',
'data': {'session': serializer.data}
})
except SessionList.DoesNotExist:
return Response({
'status': False,
'message': '会话不存在'
}, status=status.HTTP_404_NOT_FOUND)
@action(detail=False, methods=['post'])
@transaction.atomic
def update_session_status(self, request):
"""更新会话状态"""
partition_id = request.data.get('partitionId')
window_status = request.data.get('windowStatus')
window_id = request.data.get('windowId')
if not partition_id:
return Response({
'status': False,
'message': '分区ID不能为空'
}, status=status.HTTP_400_BAD_REQUEST)
try:
session = SessionList.objects.get(partitionId=partition_id)
if window_status is not None:
session.windowStatus = window_status
if window_id is not None:
session.windowId = window_id
session.save()
return Response({
'status': True,
'message': '更新成功'
})
except SessionList.DoesNotExist:
return Response({
'status': False,
'message': '会话不存在'
}, status=status.HTTP_404_NOT_FOUND)
class ContactInfoViewSet(viewsets.ModelViewSet):
"""联系人信息视图集"""
queryset = ContactInfo.objects.all()
serializer_class = ContactInfoSerializer
permission_classes = [AllowAny]
def get_queryset(self):
"""根据用户ID和平台过滤"""
queryset = ContactInfo.objects.all()
user_id = self.request.query_params.get('userId')
platform = self.request.query_params.get('platform')
if user_id:
queryset = queryset.filter(userId=user_id)
if platform:
queryset = queryset.filter(platform=platform)
return queryset.order_by('-created_at')
class FollowRecordViewSet(viewsets.ModelViewSet):
"""跟进记录视图集"""
queryset = FollowRecord.objects.all()
serializer_class = FollowRecordSerializer
permission_classes = [AllowAny]
def get_queryset(self):
"""根据用户ID和平台过滤"""
queryset = FollowRecord.objects.all()
user_id = self.request.query_params.get('userId')
platform = self.request.query_params.get('platform')
if user_id:
queryset = queryset.filter(userId=user_id)
if platform:
queryset = queryset.filter(platform=platform)
return queryset.order_by('-created_at')
class TranslateCacheViewSet(viewsets.ModelViewSet):
"""翻译缓存视图集"""
queryset = TranslateCache.objects.all()
serializer_class = TranslateCacheSerializer
permission_classes = [AllowAny]
def get_queryset(self):
"""根据分区ID过滤"""
queryset = TranslateCache.objects.all()
partition_id = self.request.query_params.get('partitionId')
if partition_id:
queryset = queryset.filter(partitionId=partition_id)
return queryset.order_by('-created_at')
@action(detail=False, methods=['post'])
def clear_cache(self, request):
"""清空翻译缓存"""
partition_id = request.data.get('partitionId')
if partition_id:
TranslateCache.objects.filter(partitionId=partition_id).delete()
return Response({
'status': True,
'message': f'已清空分区 {partition_id} 的翻译缓存'
})
else:
TranslateCache.objects.all().delete()
return Response({
'status': True,
'message': '已清空所有翻译缓存'
})
class ProxyConfigViewSet(viewsets.ModelViewSet):
"""代理配置视图集"""
queryset = ProxyConfig.objects.all()
serializer_class = ProxyConfigSerializer
permission_classes = [AllowAny]
def get_queryset(self):
"""根据分区ID过滤"""
queryset = ProxyConfig.objects.all()
partition_id = self.request.query_params.get('partitionId')
if partition_id:
queryset = queryset.filter(partitionId=partition_id)
return queryset
class GlobalProxyConfigViewSet(viewsets.ModelViewSet):
"""全局代理配置视图集"""
queryset = GlobalProxyConfig.objects.all()
serializer_class = GlobalProxyConfigSerializer
permission_classes = [AllowAny]
class GroupManageViewSet(viewsets.ModelViewSet):
"""分组管理视图集"""
queryset = GroupManage.objects.all()
serializer_class = GroupManageSerializer
permission_classes = [AllowAny]
class QuickReplyRecordViewSet(viewsets.ModelViewSet):
"""快捷回复记录视图集"""
queryset = QuickReplyRecord.objects.all()
serializer_class = QuickReplyRecordSerializer
permission_classes = [AllowAny]
def get_queryset(self):
"""根据分组ID过滤"""
queryset = QuickReplyRecord.objects.all()
group_id = self.request.query_params.get('groupId')
if group_id:
queryset = queryset.filter(groupId=group_id)
return queryset
class ParameterViewSet(viewsets.ModelViewSet):
"""参数配置视图集"""
queryset = Parameter.objects.all()
serializer_class = ParameterSerializer
permission_classes = [AllowAny]
@action(detail=False, methods=['get'])
def get_by_key(self, request):
"""根据键获取参数值"""
key = request.query_params.get('key')
if not key:
return Response({
'status': False,
'message': '参数键不能为空'
}, status=status.HTTP_400_BAD_REQUEST)
try:
parameter = Parameter.objects.get(key=key)
return Response({
'status': True,
'data': {'value': parameter.value}
})
except Parameter.DoesNotExist:
return Response({
'status': False,
'message': '参数不存在'
}, status=status.HTTP_404_NOT_FOUND)
class TgSessionsViewSet(viewsets.ModelViewSet):
"""Telegram 会话视图集"""
queryset = TgSessions.objects.all()
serializer_class = TgSessionsSerializer
permission_classes = [AllowAny]
class TranslateConfigViewSet(viewsets.ModelViewSet):
"""翻译配置视图集"""
queryset = TranslateConfig.objects.all()
serializer_class = TranslateConfigSerializer
permission_classes = [AllowAny]
@action(detail=False, methods=['get'], url_path='get_by_user_platform')
def get_by_user_platform(self, request):
"""根据用户ID和平台获取翻译配置"""
user_id = request.query_params.get('userId')
platform = request.query_params.get('platform')
try:
# 优先查找用户级配置
if user_id:
config = TranslateConfig.objects.filter(userId=user_id).first()
if config:
serializer = self.get_serializer(config)
return Response({
'status': True,
'message': '查询成功',
'data': serializer.data
})
# 查找平台级配置
if platform:
config = TranslateConfig.objects.filter(platform=platform).first()
if config:
serializer = self.get_serializer(config)
return Response({
'status': True,
'message': '查询成功',
'data': serializer.data
})
return Response({
'status': False,
'message': '配置不存在'
}, status=status.HTTP_404_NOT_FOUND)
except Exception as e:
return Response({
'status': False,
'message': f'查询失败：{str(e)}'
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
class TranslateRouteViewSet(viewsets.ModelViewSet):
"""翻译路由视图集"""
queryset = TranslateRoute.objects.all()
serializer_class = TranslateRouteSerializer
permission_classes = [AllowAny]
@action(detail=False, methods=['get'], url_path='get_by_name')
def get_by_name(self, request):
"""根据名称获取翻译路由配置"""
name = request.query_params.get('name')
if not name:
return Response({
'status': False,
'message': '缺少name参数'
}, status=status.HTTP_400_BAD_REQUEST)
try:
route = TranslateRoute.objects.filter(name=name).first()
if route:
serializer = self.get_serializer(route)
return Response({
'status': True,
'message': '查询成功',
'data': serializer.data
})
else:
return Response({
'status': False,
'message': '找不到配置信息'
}, status=status.HTTP_404_NOT_FOUND)
except Exception as e:
return Response({
'status': False,
'message': f'查询失败：{str(e)}'
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
class LanguageListViewSet(viewsets.ModelViewSet):
"""语言列表视图集"""
queryset = LanguageList.objects.all()
serializer_class = LanguageListSerializer
permission_classes = [AllowAny]

270
backend/simple_data_migration.py Normal file
View File

@ -0,0 +1,270 @@
#!/usr/bin/env python
"""
简化的数据迁移脚本 - 直接使用 SQL 语句
"""
import sqlite3
import pymysql
from pathlib import Path
import sys
def migrate_data():
"""执行数据迁移"""
print("🚀 开始数据迁移...")
# SQLite 数据库路径
sqlite_db_path = Path(__file__).parent.parent.parent / 'liangzi_data' / 'session.db'
if not sqlite_db_path.exists():
print(f"❌ SQLite 数据库文件不存在: {sqlite_db_path}")
return False
print(f"📍 SQLite 数据库路径: {sqlite_db_path}")
# MySQL 连接配置
mysql_config = {
'host': '192.168.1.114',
'port': 3306,
'user': 'seabox',
'password': '123456',
'database': 'fyapi',
'charset': 'utf8mb4'
}
try:
# 连接 SQLite
sqlite_conn = sqlite3.connect(str(sqlite_db_path))
sqlite_conn.row_factory = sqlite3.Row
sqlite_cursor = sqlite_conn.cursor()
# 连接 MySQL
mysql_conn = pymysql.connect(**mysql_config)
mysql_cursor = mysql_conn.cursor()
print("✅ 数据库连接成功")
# 定义迁移任务
migration_tasks = [
{
'name': '会话列表',
'sqlite_table': 'session_list',
'mysql_table': 'session_list',
'fields': [
'partitionId', 'windowStatus', 'createTime', 'platform',
'nickName', 'msgCount', 'onlineStatus', 'webUrl', 'windowId',
'userAgent', 'remarks', 'avatarUrl', 'userName', 'isTop'
]
},
{
'name': '联系人信息',
'sqlite_table': 'contact_info',
'mysql_table': 'contact_info',
'fields': [
'userId', 'platform', 'nickName', 'userName', 'avatarUrl',
'phoneNumber', 'email', 'remarks', 'level', 'tags', 'lastContactTime'
]
},
{
'name': '跟进记录',
'sqlite_table': 'follow_record',
'mysql_table': 'follow_record',
'fields': [
'userId', 'platform', 'content', 'followTime', 'followType',
'result', 'nextFollowTime'
]
},
{
'name': '翻译缓存',
'sqlite_table': 'translate_cache',
'mysql_table': 'translate_cache',
'fields': [
'partitionId', 'sourceText', 'translatedText', 'sourceLang',
'targetLang', 'translateService', 'cacheTime', 'hitCount'
]
},
{
'name': '代理配置',
'sqlite_table': 'proxy_config',
'mysql_table': 'proxy_config',
'fields': [
'partitionId', 'proxyStatus', 'proxyType', 'proxyIp', 'proxyPort',
'userVerifyStatus', 'username', 'password', 'timezone', 'defaultLanguage'
]
},
{
'name': '全局代理配置',
'sqlite_table': 'global_proxy_config',
'mysql_table': 'global_proxy_config',
'fields': [
'proxyStatus', 'proxyType', 'proxyIp', 'proxyPort',
'userVerifyStatus', 'username', 'password'
]
},
{
'name': '分组管理',
'sqlite_table': 'group_manage',
'mysql_table': 'group_manage',
'fields': [
'groupName', 'groupDescription', 'sortOrder', 'isActive'
]
},
{
'name': '快捷回复记录',
'sqlite_table': 'quick_reply_record',
'mysql_table': 'quick_reply_record',
'fields': [
'groupId', 'title', 'content', 'tags', 'useCount', 'sortOrder', 'isActive'
]
},
{
'name': '用户快捷回复',
'sqlite_table': 'user_quick_replies',
'mysql_table': 'user_quick_replies',
'fields': [
'userId', 'title', 'content', 'category', 'tags', 'useCount', 'isActive'
]
},
{
'name': '参数配置',
'sqlite_table': 'parameter',
'mysql_table': 'parameter',
'fields': [
'key', 'value', 'description', 'type', 'isSystem'
]
},
{
'name': 'Telegram 会话',
'sqlite_table': 'tg_sessions',
'mysql_table': 'tg_sessions',
'fields': [
'sessionId', 'userId', 'phoneNumber', 'sessionData', 'isActive', 'lastUsed'
]
}
]
total_migrated = 0
# 执行迁移
for task in migration_tasks:
print(f"\n📋 迁移 {task['name']}...")
try:
# 检查 SQLite 表是否存在
sqlite_cursor.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{task['sqlite_table']}'")
if not sqlite_cursor.fetchone():
print(f"⚠️ 表 {task['sqlite_table']} 不存在，跳过")
continue
# 获取 SQLite 数据
sqlite_cursor.execute(f"SELECT * FROM {task['sqlite_table']}")
rows = sqlite_cursor.fetchall()
if not rows:
print(f"⚠️ 表 {task['sqlite_table']} 没有数据")
continue
# 准备 MySQL 插入语句
fields_str = ', '.join([f"`{field}`" for field in task['fields']])
placeholders = ', '.join(['%s'] * len(task['fields']))
# 检查是否有重复数据（针对有唯一约束的表）
if task['sqlite_table'] == 'session_list':
# 清空现有数据以避免重复
mysql_cursor.execute(f"DELETE FROM {task['mysql_table']} WHERE partitionId IN (SELECT partitionId FROM (SELECT partitionId FROM {task['mysql_table']}) AS temp)")
insert_sql = f"INSERT INTO {task['mysql_table']} ({fields_str}) VALUES ({placeholders})"
migrated_count = 0
for row in rows:
try:
# 提取字段值
values = []
for field in task['fields']:
value = row[field] if field in row.keys() else None
values.append(value)
# 插入 MySQL
mysql_cursor.execute(insert_sql, values)
migrated_count += 1
except Exception as e:
print(f"❌ 插入数据失败: {e}")
continue
mysql_conn.commit()
print(f"✅ {task['name']} 迁移完成，共 {migrated_count} 条记录")
total_migrated += migrated_count
except Exception as e:
print(f"❌ 迁移 {task['name']} 失败: {e}")
continue
# 关闭连接
sqlite_conn.close()
mysql_conn.close()
print(f"\n🎉 数据迁移完成！总共迁移了 {total_migrated} 条记录")
# 验证迁移结果
print("\n🔍 验证迁移结果...")
verify_migration()
return True
except Exception as e:
print(f"❌ 迁移过程中发生错误: {e}")
return False
def verify_migration():
"""验证迁移结果"""
mysql_config = {
'host': '192.168.1.114',
'port': 3306,
'user': 'seabox',
'password': '123456',
'database': 'fyapi',
'charset': 'utf8mb4'
}
try:
mysql_conn = pymysql.connect(**mysql_config)
mysql_cursor = mysql_conn.cursor()
tables_to_check = [
'session_list', 'contact_info', 'follow_record', 'translate_cache',
'proxy_config', 'global_proxy_config', 'group_manage',
'quick_reply_record', 'user_quick_replies', 'parameter', 'tg_sessions'
]
total_records = 0
for table in tables_to_check:
try:
mysql_cursor.execute(f"SELECT COUNT(*) FROM {table}")
count = mysql_cursor.fetchone()[0]
print(f"{table}: {count} 条记录")
total_records += count
except Exception as e:
print(f"{table}: 查询失败 - {e}")
print(f"\n📊 MySQL 数据库总记录数: {total_records} 条")
mysql_conn.close()
except Exception as e:
print(f"❌ 验证失败: {e}")
if __name__ == "__main__":
print("=" * 60)
print("SQLite 到 MySQL 数据迁移工具")
print("=" * 60)
success = migrate_data()
if success:
print("\n✅ 迁移成功完成！")
print("\n下一步:")
print("1. 启动 Django 服务: python manage.py runserver")
print("2. 测试 API 接口")
print("3. 启动前端应用")
else:
print("\n❌ 迁移失败，请检查错误信息")
sys.exit(1)

25
backend/simple_test.py Normal file
View File

@ -0,0 +1,25 @@
print("Hello World!")
print("Python is working")
import sys
print(f"Python version: {sys.version}")
try:
import django
print(f"Django version: {django.get_version()}")
except ImportError as e:
print(f"Django import error: {e}")
try:
import sqlite3
print("SQLite3 module available")
except ImportError as e:
print(f"SQLite3 import error: {e}")
try:
import pymysql
print("PyMySQL module available")
except ImportError as e:
print(f"PyMySQL import error: {e}")
print("Test completed!")

104
backend/test_migration.py Normal file
View File

@ -0,0 +1,104 @@
#!/usr/bin/env python
"""
测试迁移脚本
"""
import os
import sys
import sqlite3
import django
from pathlib import Path
print("🚀 开始测试迁移...")
# 设置Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
try:
print("📋 设置Django环境...")
django.setup()
print("✅ Django环境设置成功")
# 导入模型
from session.models import SessionList
print("✅ 模型导入成功")
# 检查SQLite数据库
sqlite_db_path = Path(__file__).parent.parent.parent / 'liangzi_data' / 'session.db'
print(f"📍 SQLite 数据库路径: {sqlite_db_path}")
if not sqlite_db_path.exists():
print(f"❌ SQLite 数据库文件不存在: {sqlite_db_path}")
sys.exit(1)
print("✅ SQLite 数据库文件存在")
# 连接SQLite数据库
sqlite_conn = sqlite3.connect(str(sqlite_db_path))
sqlite_conn.row_factory = sqlite3.Row
cursor = sqlite_conn.cursor()
# 检查表是否存在
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = cursor.fetchall()
print(f"📋 SQLite 数据库中的表: {[table[0] for table in tables]}")
# 检查session_list表的数据
try:
cursor.execute("SELECT COUNT(*) FROM session_list")
count = cursor.fetchone()[0]
print(f"📊 session_list 表中有 {count} 条记录")
if count > 0:
cursor.execute("SELECT * FROM session_list LIMIT 3")
rows = cursor.fetchall()
print("📋 前3条记录:")
for row in rows:
print(f" - partitionId: {row['partitionId']}, platform: {row['platform']}")
except Exception as e:
print(f"❌ 查询session_list表失败: {e}")
# 测试MySQL连接
try:
import pymysql
print("✅ pymysql 模块导入成功")
# 测试Django数据库连接
from django.db import connection
cursor = connection.cursor()
cursor.execute("SELECT 1")
result = cursor.fetchone()
print(f"✅ Django MySQL 连接测试成功: {result}")
# 检查MySQL中的表
cursor.execute("SHOW TABLES")
mysql_tables = cursor.fetchall()
print(f"📋 MySQL 数据库中的表: {[table[0] for table in mysql_tables]}")
# 检查session_list表是否存在
cursor.execute("SHOW TABLES LIKE 'session_list'")
session_table = cursor.fetchone()
if session_table:
print("✅ session_list 表已存在于MySQL中")
# 检查现有数据
cursor.execute("SELECT COUNT(*) FROM session_list")
mysql_count = cursor.fetchone()[0]
print(f"📊 MySQL session_list 表中有 {mysql_count} 条记录")
else:
print("⚠️ session_list 表不存在于MySQL中")
except Exception as e:
print(f"❌ MySQL 连接测试失败: {e}")
import traceback
traceback.print_exc()
sqlite_conn.close()
print("✅ 测试完成")
except Exception as e:
print(f"❌ 测试过程中发生错误: {e}")
import traceback
traceback.print_exc()
sys.exit(1)

283
backend/verify_migration.py Normal file
View File

@ -0,0 +1,283 @@
#!/usr/bin/env python
"""
迁移验证脚本 - 验证所有功能是否正常工作
"""
import os
import sys
import django
import requests
import json
from pathlib import Path
# 设置Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
def test_django_setup():
"""测试Django环境设置"""
print("🔧 测试Django环境...")
try:
django.setup()
print("✅ Django环境设置成功")
return True
except Exception as e:
print(f"❌ Django环境设置失败: {e}")
return False
def test_models():
"""测试模型导入"""
print("📋 测试模型导入...")
try:
from session.models import SessionList, ContactInfo, ProxyConfig
from translate.models import TranslateConfig
print("✅ 模型导入成功")
return True
except Exception as e:
print(f"❌ 模型导入失败: {e}")
return False
def test_database_connection():
"""测试数据库连接"""
print("🗄️ 测试数据库连接...")
try:
from django.db import connection
cursor = connection.cursor()
cursor.execute("SELECT 1")
result = cursor.fetchone()
print(f"✅ 数据库连接成功: {result}")
return True
except Exception as e:
print(f"❌ 数据库连接失败: {e}")
return False
def test_tables_exist():
"""测试表是否存在"""
print("📊 测试表结构...")
try:
from django.db import connection
cursor = connection.cursor()
# 检查重要表是否存在
tables_to_check = [
'session_list', 'translate_config', 'contact_info',
'proxy_config', 'translate_cache', 'parameter'
]
existing_tables = []
for table in tables_to_check:
cursor.execute(f"SHOW TABLES LIKE '{table}'")
if cursor.fetchone():
existing_tables.append(table)
print(f"✅ 存在的表: {existing_tables}")
print(f"📊 总计: {len(existing_tables)}/{len(tables_to_check)} 个表")
return len(existing_tables) > 0
except Exception as e:
print(f"❌ 表结构检查失败: {e}")
return False
def test_api_endpoints():
"""测试API接口"""
print("🌐 测试API接口...")
base_url = "http://localhost:8000"
endpoints_to_test = [
"/api/session/sessions/",
"/api/session/contacts/",
"/api/session/proxy-config/",
"/api/session/global-proxy-config/",
"/api/session/parameters/",
]
working_endpoints = []
for endpoint in endpoints_to_test:
try:
response = requests.get(f"{base_url}{endpoint}", timeout=5)
if response.status_code in [200, 404]: # 404也算正常，说明路由存在
working_endpoints.append(endpoint)
print(f"✅ {endpoint} - 状态码: {response.status_code}")
else:
print(f"⚠️ {endpoint} - 状态码: {response.status_code}")
except requests.exceptions.ConnectionError:
print(f"❌ {endpoint} - 连接失败（服务可能未启动）")
except Exception as e:
print(f"❌ {endpoint} - 错误: {e}")
print(f"📊 API测试结果: {len(working_endpoints)}/{len(endpoints_to_test)} 个接口可访问")
return len(working_endpoints) > 0
def test_session_api():
"""测试会话API功能"""
print("🔧 测试会话API功能...")
base_url = "http://localhost:8000"
try:
# 测试添加会话
add_session_data = {
"platform": "TestPlatform",
"url": "https://test.com",
"nickname": "测试会话"
}
response = requests.post(
f"{base_url}/api/session/sessions/add_session/",
json=add_session_data,
timeout=10
)
if response.status_code == 200:
result = response.json()
if result.get('status'):
partition_id = result.get('data', {}).get('partitionId')
print(f"✅ 添加会话成功，分区ID: {partition_id}")
# 测试获取会话
get_response = requests.get(
f"{base_url}/api/session/sessions/get_by_partition_id/",
params={"partitionId": partition_id},
timeout=10
)
if get_response.status_code == 200:
print("✅ 获取会话信息成功")
return True
else:
print(f"⚠️ 获取会话信息失败: {get_response.status_code}")
else:
print(f"⚠️ 添加会话失败: {result.get('message')}")
else:
print(f"⚠️ 添加会话请求失败: {response.status_code}")
except requests.exceptions.ConnectionError:
print("❌ API测试失败：无法连接到服务器（请确保后端服务已启动）")
except Exception as e:
print(f"❌ API测试失败: {e}")
return False
def generate_verification_report():
"""生成验证报告"""
print("\n" + "=" * 80)
print("📋 生成迁移验证报告...")
report_content = f"""# SQLite 到 MySQL 迁移验证报告
## 验证时间
- 验证日期: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
## 验证结果
### 1. Django环境
- Django设置: {'✅ 成功' if test_django_setup() else '❌ 失败'}
- 模型导入: {'✅ 成功' if test_models() else '❌ 失败'}
### 2. 数据库
- 数据库连接: {'✅ 成功' if test_database_connection() else '❌ 失败'}
- 表结构检查: {'✅ 成功' if test_tables_exist() else '❌ 失败'}
### 3. API接口
- 接口可访问性: {'✅ 成功' if test_api_endpoints() else '❌ 失败'}
- 会话API功能: {'✅ 成功' if test_session_api() else '❌ 失败'}
## 建议
### 如果验证失败：
1. **Django环境问题**
```bash
pip install -r requirements.txt
python manage.py check
```
2. **数据库连接问题**
```bash
python manage.py migrate
python manage.py dbshell
```
3. **API接口问题**
```bash
python manage.py runserver 0.0.0.0:8000
```
### 如果验证成功：
1. 可以开始使用新的API接口
2. 前端应用可以连接到后端服务
3. 数据迁移已完成
## 下一步
1. 启动前端应用
2. 测试完整的用户流程
3. 监控系统运行状态
4. 定期备份数据
验证完成！
"""
# 保存报告
report_file = Path(__file__).parent / f'verification_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.md'
with open(report_file, 'w', encoding='utf-8') as f:
f.write(report_content)
print(f"📄 验证报告已保存到: {report_file}")
def main():
"""主函数"""
print("🚀 开始迁移验证...")
print("=" * 80)
# 导入datetime
from datetime import datetime
# 执行所有测试
tests = [
("Django环境设置", test_django_setup),
("模型导入", test_models),
("数据库连接", test_database_connection),
("表结构检查", test_tables_exist),
("API接口测试", test_api_endpoints),
("会话API功能", test_session_api),
]
results = []
for test_name, test_func in tests:
print(f"\n🔍 执行测试: {test_name}")
try:
result = test_func()
results.append((test_name, result))
except Exception as e:
print(f"❌ 测试 {test_name} 出现异常: {e}")
results.append((test_name, False))
# 输出总结
print("\n" + "=" * 80)
print("📊 验证结果总结:")
print("-" * 80)
passed = 0
total = len(results)
for test_name, result in results:
status = "✅ 通过" if result else "❌ 失败"
print(f"{test_name}: {status}")
if result:
passed += 1
print("-" * 80)
print(f"总计: {passed}/{total} 项测试通过")
if passed == total:
print("🎉 所有测试通过！迁移验证成功！")
elif passed > total // 2:
print("⚠️ 部分测试通过，请检查失败的项目")
else:
print("❌ 多项测试失败，请检查系统配置")
print("\n✅ 验证完成！")
if __name__ == "__main__":
main()