290 lines
11 KiB
Python
290 lines
11 KiB
Python
# -*- coding: UTF-8 -*-
|
||
'''
|
||
@Project :domainScanDemo
|
||
@File :export_manager.py
|
||
@IDE :PyCharm
|
||
@Author :梦伴
|
||
@Date :2026/4/8 23:54
|
||
@explain : 导出管理器
|
||
'''
|
||
|
||
import os
|
||
from loguru import logger
|
||
from app.utils.database import Database
|
||
|
||
|
||
class ExportManager:
|
||
"""
|
||
导出管理器
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""
|
||
初始化导出管理器
|
||
"""
|
||
self.db = Database()
|
||
|
||
def export_domains(self, filter_conditions, output_file):
|
||
"""
|
||
导出域名
|
||
|
||
:param filter_conditions: 筛选条件
|
||
:param output_file: 输出文件路径
|
||
:return: int - 导出的域名数量
|
||
"""
|
||
try:
|
||
# 查询符合条件的域名
|
||
domains = self.db.get_domains_by_conditions(filter_conditions)
|
||
|
||
if not domains:
|
||
logger.warning("没有符合条件的域名")
|
||
return 0
|
||
|
||
# 导出到文件
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
for domain in domains:
|
||
f.write(domain['domain'] + '\n')
|
||
|
||
logger.info(f"成功导出 {len(domains)} 个域名到 {output_file}")
|
||
return len(domains)
|
||
except Exception as e:
|
||
logger.error(f"导出域名出错: {e}")
|
||
return 0
|
||
|
||
def export_with_details(self, filter_conditions, output_file):
|
||
"""
|
||
导出域名及其详细信息
|
||
|
||
:param filter_conditions: 筛选条件
|
||
:param output_file: 输出文件路径
|
||
:return: int - 导出的域名数量
|
||
"""
|
||
try:
|
||
# 查询符合条件的域名及其详细信息
|
||
domains = self.db.get_domains_with_details(filter_conditions)
|
||
|
||
if not domains:
|
||
logger.warning("没有符合条件的域名")
|
||
return 0
|
||
|
||
# 导出到文件
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
# 写入表头
|
||
f.write('域名,注册状态,使用状态,检测状态,备案历史,备案年份,快照年份,友情链接数量\n')
|
||
|
||
# 写入数据
|
||
for domain in domains:
|
||
line = f"{domain['domain']},{domain['register_status']},{domain['use_status']},{domain['detect_status']},{domain['has_beian']},{domain['beian_year']},{domain['snapshot_years']},{domain['backlink_count']}\n"
|
||
f.write(line)
|
||
|
||
logger.info(f"成功导出 {len(domains)} 个域名及其详细信息到 {output_file}")
|
||
return len(domains)
|
||
except Exception as e:
|
||
logger.error(f"导出域名详细信息出错: {e}")
|
||
return 0
|
||
|
||
def batch_update_status(self, domain_ids, status_type, status_value):
|
||
"""
|
||
批量更新域名状态
|
||
|
||
:param domain_ids: 域名ID列表
|
||
:param status_type: 状态类型
|
||
:param status_value: 状态值
|
||
:return: int - 更新成功的域名数量
|
||
"""
|
||
try:
|
||
success_count = 0
|
||
for domain_id in domain_ids:
|
||
if self.db.update_domain_status(domain_id, status_type, status_value):
|
||
success_count += 1
|
||
|
||
logger.info(f"成功更新 {success_count} 个域名的状态")
|
||
return success_count
|
||
except Exception as e:
|
||
logger.error(f"批量更新域名状态出错: {e}")
|
||
return 0
|
||
|
||
def get_filtered_domains(self, filter_conditions, limit=1000):
|
||
"""
|
||
获取符合条件的域名
|
||
|
||
:param filter_conditions: 筛选条件
|
||
:param limit: 限制数量
|
||
:return: list - 域名列表
|
||
"""
|
||
try:
|
||
domains = self.db.get_domains_by_conditions(filter_conditions, limit)
|
||
logger.info(f"获取到 {len(domains)} 个符合条件的域名")
|
||
return domains
|
||
except Exception as e:
|
||
logger.error(f"获取符合条件的域名出错: {e}")
|
||
return []
|
||
|
||
def generate_report(self, output_file):
|
||
"""
|
||
生成统计报告
|
||
|
||
:param output_file: 输出文件路径
|
||
:return: bool - 是否生成成功
|
||
"""
|
||
try:
|
||
# 获取域名统计信息
|
||
domain_stats = self.db.get_domain_statistics()
|
||
|
||
# 获取任务统计信息
|
||
task_stats = self.db.get_task_statistics()
|
||
|
||
# 生成报告
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
f.write('域名库系统统计报告\n')
|
||
f.write('=' * 50 + '\n')
|
||
|
||
f.write('\n域名统计:\n')
|
||
f.write(f'总域名数: {domain_stats.get("total", 0)}\n')
|
||
f.write(f'可注册域名: {domain_stats.get("available", 0)}\n')
|
||
f.write(f'已注册域名: {domain_stats.get("registered", 0)}\n')
|
||
f.write(f'黑名单域名: {domain_stats.get("blacklisted", 0)}\n')
|
||
|
||
f.write('\n任务统计:\n')
|
||
f.write(f'总任务数: {task_stats.get("total", 0)}\n')
|
||
f.write(f'待执行任务: {task_stats.get("pending", 0)}\n')
|
||
f.write(f'执行中任务: {task_stats.get("running", 0)}\n')
|
||
f.write(f'已完成任务: {task_stats.get("completed", 0)}\n')
|
||
f.write(f'失败任务: {task_stats.get("failed", 0)}\n')
|
||
|
||
logger.info(f"成功生成统计报告到 {output_file}")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"生成统计报告出错: {e}")
|
||
return False
|
||
|
||
def export_to_excel(self, domains, output_file):
|
||
"""
|
||
导出域名到Excel文件
|
||
|
||
:param domains: 域名列表
|
||
:param output_file: 输出文件路径
|
||
:return: int - 导出的域名数量
|
||
"""
|
||
try:
|
||
# 尝试导入openpyxl
|
||
try:
|
||
from openpyxl import Workbook
|
||
except ImportError:
|
||
# 如果没有安装openpyxl，使用CSV格式作为替代
|
||
logger.warning("openpyxl库未安装，将使用CSV格式导出")
|
||
return self.export_to_csv(domains, output_file.replace('.xlsx', '.csv'))
|
||
|
||
# 创建工作簿
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
|
||
# 写入表头
|
||
headers = ['域名', '注册状态', '使用状态', '检测状态', '人工复核状态', '过期时间', '单位性质', '网站首页网址', '检测时间', '备案历史', '备案年份', '快照年份', '百度历史', '百度Site', '是否中文标题', '360 Site', 'Google Site', '友情链接数量']
|
||
ws.append(headers)
|
||
|
||
# 写入数据
|
||
for domain in domains:
|
||
row = [
|
||
domain.get('domain', ''),
|
||
domain.get('register_status', ''),
|
||
domain.get('use_status', ''),
|
||
domain.get('detect_status', ''),
|
||
domain.get('review_status', ''),
|
||
domain.get('expire_date', ''),
|
||
domain.get('company_type', ''),
|
||
domain.get('website_url', ''),
|
||
domain.get('detect_time', ''),
|
||
domain.get('has_beian', ''),
|
||
domain.get('beian_year', ''),
|
||
domain.get('snapshot_years', ''),
|
||
domain.get('baidu_history', ''),
|
||
domain.get('baidu_site', ''),
|
||
domain.get('is_chinese_title', ''),
|
||
domain.get('qihu360_site', ''),
|
||
domain.get('google_site', ''),
|
||
domain.get('backlink_count', '')
|
||
]
|
||
ws.append(row)
|
||
|
||
# 保存文件
|
||
wb.save(output_file)
|
||
|
||
logger.info(f"成功导出 {len(domains)} 个域名到Excel文件: {output_file}")
|
||
return len(domains)
|
||
except Exception as e:
|
||
logger.error(f"导出Excel文件出错: {e}")
|
||
# 尝试使用CSV格式作为替代
|
||
try:
|
||
csv_file = output_file.replace('.xlsx', '.csv')
|
||
logger.info(f"尝试使用CSV格式导出到: {csv_file}")
|
||
return self.export_to_csv(domains, csv_file)
|
||
except Exception as e2:
|
||
logger.error(f"导出CSV文件也失败: {e2}")
|
||
raise
|
||
|
||
def export_to_csv(self, domains, output_file):
|
||
"""
|
||
导出域名到CSV文件
|
||
|
||
:param domains: 域名列表
|
||
:param output_file: 输出文件路径
|
||
:return: int - 导出的域名数量
|
||
"""
|
||
try:
|
||
import csv
|
||
|
||
# 写入文件
|
||
with open(output_file, 'w', encoding='utf-8', newline='') as f:
|
||
writer = csv.writer(f)
|
||
|
||
# 写入表头
|
||
headers = ['域名', '注册状态', '使用状态', '检测状态', '人工复核状态', '过期时间', '单位性质', '网站首页网址', '检测时间', '备案历史', '备案年份', '快照年份', '百度历史', '百度Site', '是否中文标题', '360 Site', 'Google Site', '友情链接数量']
|
||
writer.writerow(headers)
|
||
|
||
# 写入数据
|
||
for domain in domains:
|
||
row = [
|
||
domain.get('domain', ''),
|
||
domain.get('register_status', ''),
|
||
domain.get('use_status', ''),
|
||
domain.get('detect_status', ''),
|
||
domain.get('review_status', ''),
|
||
domain.get('expire_date', ''),
|
||
domain.get('company_type', ''),
|
||
domain.get('website_url', ''),
|
||
domain.get('detect_time', ''),
|
||
domain.get('has_beian', ''),
|
||
domain.get('beian_year', ''),
|
||
domain.get('snapshot_years', ''),
|
||
domain.get('baidu_history', ''),
|
||
domain.get('baidu_site', ''),
|
||
domain.get('is_chinese_title', ''),
|
||
domain.get('qihu360_site', ''),
|
||
domain.get('google_site', ''),
|
||
domain.get('backlink_count', '')
|
||
]
|
||
writer.writerow(row)
|
||
|
||
logger.info(f"成功导出 {len(domains)} 个域名到CSV文件: {output_file}")
|
||
return len(domains)
|
||
except Exception as e:
|
||
logger.error(f"导出CSV文件出错: {e}")
|
||
raise
|
||
|
||
def export_to_txt(self, domains, output_file):
|
||
"""
|
||
导出域名到TXT文件，一行一个域名。
|
||
"""
|
||
try:
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
for domain in domains:
|
||
value = domain.get('domain', '').strip()
|
||
if value:
|
||
f.write(value + '\n')
|
||
logger.info(f"成功导出 {len(domains)} 个域名到TXT文件: {output_file}")
|
||
return len(domains)
|
||
except Exception as e:
|
||
logger.error(f"导出TXT文件出错: {e}")
|
||
raise
|