770 lines
32 KiB
Python
770 lines
32 KiB
Python
import os
|
||
import shutil
|
||
import sys
|
||
import uuid
|
||
import logging
|
||
import threading
|
||
import time
|
||
from functools import wraps
|
||
from pathlib import Path
|
||
from flask import Flask, request, jsonify, send_file, render_template_string, url_for, g
|
||
from flask_cors import CORS
|
||
from werkzeug.utils import secure_filename
|
||
import yaml
|
||
|
||
# Shared utilities imported from shared.py
|
||
try:
|
||
# Try relative import (when run as part of package)
|
||
from .shared import TASK_STATUS_PENDING, TASK_STATUS_PROCESSING, TASK_STATUS_COMPLETED, TASK_STATUS_FAILED, update_task_status as shared_update_task_status
|
||
except ImportError:
|
||
# Fallback to absolute import (when run directly)
|
||
from shared import TASK_STATUS_PENDING, TASK_STATUS_PROCESSING, TASK_STATUS_COMPLETED, TASK_STATUS_FAILED, update_task_status as shared_update_task_status
|
||
|
||
# Load configuration from INI file
|
||
try:
|
||
from .config_reader import config_reader
|
||
except ImportError:
|
||
from config_reader import config_reader
|
||
|
||
# Blueprints will be imported after app initialization to avoid circular imports
|
||
|
||
# Environment-based configuration management
|
||
class Config:
|
||
"""Configuration management using INI file with environment variable fallbacks."""
|
||
|
||
# Server configuration from config_reader
|
||
HOST = config_reader.host
|
||
PORT = config_reader.port
|
||
DEBUG = config_reader.debug
|
||
BASE_URL = config_reader.base_url
|
||
|
||
# Directory configuration
|
||
BASE_DIR = None # Will be set dynamically
|
||
UPLOAD_FOLDER_NAME = str(config_reader.uploads_path)
|
||
OUTPUT_FOLDER_NAME = str(config_reader.outputs_path)
|
||
|
||
# File size limits (in bytes)
|
||
MAX_CONTENT_LENGTH = config_reader.max_content_length
|
||
|
||
# Logging configuration
|
||
LOG_LEVEL = config_reader.log_level.upper()
|
||
LOG_FILE = config_reader.log_file
|
||
|
||
# CORS configuration (keeping environment fallback for now)
|
||
CORS_ORIGINS = os.getenv('GASFLUX_CORS_ORIGINS', '*').split(',')
|
||
|
||
# Task management
|
||
TASK_CLEANUP_INTERVAL = config_reader.task_cleanup_interval
|
||
MAX_TASK_AGE = config_reader.max_task_age
|
||
SUCCESSFUL_TASK_CLEANUP_AGE = config_reader.successful_task_cleanup_age
|
||
FAILED_TASK_CLEANUP_AGE = config_reader.failed_task_cleanup_age
|
||
|
||
# Performance tuning
|
||
THREADS = config_reader.threads
|
||
CONNECTION_LIMIT = config_reader.connection_limit
|
||
CHANNEL_TIMEOUT = config_reader.channel_timeout
|
||
|
||
# Database configuration
|
||
DB_PATH = config_reader.db_path if config_reader.db_path else None
|
||
|
||
# Persistence backend
|
||
TASK_PERSIST_BACKEND = config_reader.persist_backend
|
||
|
||
# Janitor configuration
|
||
JANITOR_DRY_RUN = config_reader.janitor_dry_run
|
||
|
||
# Admin bootstrap key
|
||
ADMIN_BOOTSTRAP_KEY = config_reader.admin_bootstrap_key
|
||
|
||
@classmethod
|
||
def init_base_dir(cls):
|
||
"""Initialize base directory based on environment."""
|
||
try:
|
||
if getattr(sys, 'frozen', False):
|
||
# Running in PyInstaller bundle
|
||
cls.BASE_DIR = Path(sys.executable).parent
|
||
else:
|
||
# Running in normal Python environment
|
||
cls.BASE_DIR = Path(__file__).resolve().parent.parent.parent
|
||
except:
|
||
# Fallback to current working directory
|
||
cls.BASE_DIR = Path.cwd()
|
||
|
||
# Initialize directories based on config
|
||
cls.init_directories()
|
||
|
||
@classmethod
|
||
def init_directories(cls, output_dir=None):
|
||
"""Initialize upload and output directories from configuration."""
|
||
# Use paths from config_reader
|
||
uploads_path = config_reader.uploads_path
|
||
outputs_path = config_reader.outputs_path
|
||
|
||
# Resolve relative paths to absolute if needed
|
||
if not uploads_path.is_absolute():
|
||
uploads_path = cls.BASE_DIR / uploads_path
|
||
if not outputs_path.is_absolute():
|
||
outputs_path = cls.BASE_DIR / outputs_path
|
||
|
||
# Set the resolved paths
|
||
cls.UPLOAD_FOLDER = uploads_path
|
||
cls.OUTPUT_FOLDER = outputs_path
|
||
|
||
# Create directories
|
||
cls.UPLOAD_FOLDER.mkdir(parents=True, exist_ok=True)
|
||
cls.OUTPUT_FOLDER.mkdir(parents=True, exist_ok=True)
|
||
logger.info(f"Directories initialized - Upload: {cls.UPLOAD_FOLDER}, Output: {cls.OUTPUT_FOLDER}")
|
||
|
||
# For backward compatibility, also set the old-style paths
|
||
if output_dir:
|
||
logger.warning("output_dir parameter is deprecated, use gasflux.ini [paths] section instead")
|
||
|
||
@classmethod
|
||
def update_directories_from_config(cls, config_path=None):
|
||
"""Update directories based on config file. (DEPRECATED: Use gasflux.ini instead)"""
|
||
logger.warning("update_directories_from_config is deprecated. Output directories are now configured via gasflux.ini [paths] section.")
|
||
# No longer reads output_dir from YAML config - directories are set from INI config in init_directories()
|
||
|
||
@classmethod
|
||
def get_log_level(cls):
|
||
"""Get logging level from string."""
|
||
levels = {
|
||
'DEBUG': logging.DEBUG,
|
||
'INFO': logging.INFO,
|
||
'WARNING': logging.WARNING,
|
||
'ERROR': logging.ERROR,
|
||
'CRITICAL': logging.CRITICAL
|
||
}
|
||
return levels.get(cls.LOG_LEVEL, logging.INFO)
|
||
|
||
@classmethod
|
||
def to_dict(cls):
|
||
"""Return configuration as dictionary for debugging."""
|
||
return {
|
||
'host': cls.HOST,
|
||
'port': cls.PORT,
|
||
'debug': cls.DEBUG,
|
||
'base_dir': str(cls.BASE_DIR) if cls.BASE_DIR else None,
|
||
'upload_folder': str(cls.UPLOAD_FOLDER) if hasattr(cls, 'UPLOAD_FOLDER') else None,
|
||
'output_folder': str(cls.OUTPUT_FOLDER) if hasattr(cls, 'OUTPUT_FOLDER') else None,
|
||
'max_content_length': cls.MAX_CONTENT_LENGTH,
|
||
'log_level': cls.LOG_LEVEL,
|
||
'log_file': cls.LOG_FILE,
|
||
'cors_origins': cls.CORS_ORIGINS,
|
||
'task_cleanup_interval': cls.TASK_CLEANUP_INTERVAL,
|
||
'max_task_age': cls.MAX_TASK_AGE,
|
||
'successful_task_cleanup_age': cls.SUCCESSFUL_TASK_CLEANUP_AGE,
|
||
'failed_task_cleanup_age': cls.FAILED_TASK_CLEANUP_AGE,
|
||
'threads': cls.THREADS,
|
||
'connection_limit': cls.CONNECTION_LIMIT,
|
||
'channel_timeout': cls.CHANNEL_TIMEOUT
|
||
}
|
||
|
||
# Initialize logging with environment-based configuration
|
||
logging.basicConfig(
|
||
level=Config.get_log_level(),
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||
handlers=[
|
||
logging.StreamHandler(), # Console output
|
||
]
|
||
)
|
||
logger = logging.getLogger("gasflux_api")
|
||
logger.info("Basic logging initialized")
|
||
|
||
|
||
def log_performance(func):
|
||
"""Decorator to log function performance."""
|
||
@wraps(func)
|
||
def wrapper(*args, **kwargs):
|
||
start_time = time.time()
|
||
func_name = func.__name__
|
||
logger.debug(f"PERF: Starting {func_name}")
|
||
try:
|
||
result = func(*args, **kwargs)
|
||
duration = time.time() - start_time
|
||
logger.info(f"PERF: {func_name} completed in {duration:.3f}s")
|
||
return result
|
||
except Exception as e:
|
||
duration = time.time() - start_time
|
||
logger.error(f"PERF: {func_name} failed after {duration:.3f}s - Error: {str(e)}")
|
||
raise
|
||
return wrapper
|
||
|
||
# Task status management
|
||
# Task status constants and storage moved to shared.py
|
||
|
||
def update_task_status(task_id, status, message=None, results=None, error=None, output_dir=None):
|
||
"""Update task status using the shared implementation (writes to SQLite)."""
|
||
return shared_update_task_status(task_id, status, message=message, results=results, error=error, output_dir=output_dir)
|
||
|
||
|
||
# Statistics and Monitoring
|
||
class APIStatsCollector:
|
||
"""Collect and manage API statistics."""
|
||
|
||
def __init__(self):
|
||
self.start_time = time.time()
|
||
self.reset_stats()
|
||
|
||
def reset_stats(self):
|
||
"""Reset all statistics."""
|
||
self.stats = {
|
||
'requests': {
|
||
'total': 0,
|
||
'by_method': {},
|
||
'by_endpoint': {},
|
||
'by_status': {},
|
||
'response_times': [],
|
||
'errors': 0
|
||
},
|
||
'tasks': {
|
||
'total_created': 0,
|
||
'total_completed': 0,
|
||
'total_failed': 0,
|
||
'by_status': {
|
||
'pending': 0,
|
||
'processing': 0,
|
||
'completed': 0,
|
||
'failed': 0
|
||
},
|
||
'processing_times': []
|
||
},
|
||
'performance': {
|
||
'avg_response_time': 0,
|
||
'max_response_time': 0,
|
||
'min_response_time': float('inf'),
|
||
'uptime_seconds': time.time() - self.start_time
|
||
}
|
||
}
|
||
|
||
def record_request(self, method, endpoint, status_code, response_time):
|
||
"""Record an API request."""
|
||
self.stats['requests']['total'] += 1
|
||
|
||
# Method stats
|
||
if method not in self.stats['requests']['by_method']:
|
||
self.stats['requests']['by_method'][method] = 0
|
||
self.stats['requests']['by_method'][method] += 1
|
||
|
||
# Endpoint stats
|
||
if endpoint not in self.stats['requests']['by_endpoint']:
|
||
self.stats['requests']['by_endpoint'][endpoint] = 0
|
||
self.stats['requests']['by_endpoint'][endpoint] += 1
|
||
|
||
# Status stats
|
||
status_category = str(status_code // 100 * 100) # 200, 400, 500, etc.
|
||
if status_category not in self.stats['requests']['by_status']:
|
||
self.stats['requests']['by_status'][status_category] = 0
|
||
self.stats['requests']['by_status'][status_category] += 1
|
||
|
||
# Response time stats
|
||
self.stats['requests']['response_times'].append(response_time)
|
||
|
||
# Keep only last 1000 response times for memory efficiency
|
||
if len(self.stats['requests']['response_times']) > 1000:
|
||
self.stats['requests']['response_times'] = self.stats['requests']['response_times'][-1000:]
|
||
|
||
# Error tracking
|
||
if status_code >= 400:
|
||
self.stats['requests']['errors'] += 1
|
||
|
||
# Update performance stats
|
||
self._update_performance_stats()
|
||
|
||
def record_task_status_change(self, old_status, new_status):
|
||
"""Record task status changes."""
|
||
if old_status == "unknown": # New task
|
||
self.stats['tasks']['total_created'] += 1
|
||
|
||
if new_status == TASK_STATUS_COMPLETED:
|
||
self.stats['tasks']['total_completed'] += 1
|
||
elif new_status == TASK_STATUS_FAILED:
|
||
self.stats['tasks']['total_failed'] += 1
|
||
|
||
# Update status counts
|
||
for status in [old_status, new_status]:
|
||
if status in self.stats['tasks']['by_status']:
|
||
if status == old_status and old_status != "unknown":
|
||
self.stats['tasks']['by_status'][old_status] -= 1
|
||
elif status == new_status:
|
||
self.stats['tasks']['by_status'][new_status] += 1
|
||
|
||
def record_task_completion_time(self, completion_time):
|
||
"""Record task completion time."""
|
||
self.stats['tasks']['processing_times'].append(completion_time)
|
||
|
||
# Keep only last 100 processing times
|
||
if len(self.stats['tasks']['processing_times']) > 100:
|
||
self.stats['tasks']['processing_times'] = self.stats['tasks']['processing_times'][-100:]
|
||
|
||
def _update_performance_stats(self):
|
||
"""Update performance statistics."""
|
||
response_times = self.stats['requests']['response_times']
|
||
if response_times:
|
||
self.stats['performance']['avg_response_time'] = sum(response_times) / len(response_times)
|
||
self.stats['performance']['max_response_time'] = max(response_times)
|
||
self.stats['performance']['min_response_time'] = min(response_times)
|
||
|
||
self.stats['performance']['uptime_seconds'] = time.time() - self.start_time
|
||
|
||
def get_summary(self):
|
||
"""Get a summary of current statistics."""
|
||
current_time = time.time()
|
||
uptime = current_time - self.start_time
|
||
|
||
# Calculate rates
|
||
requests_per_second = self.stats['requests']['total'] / max(uptime, 1)
|
||
error_rate = (self.stats['requests']['errors'] / max(self.stats['requests']['total'], 1)) * 100
|
||
|
||
# Task completion rate
|
||
total_tasks_processed = self.stats['tasks']['total_completed'] + self.stats['tasks']['total_failed']
|
||
task_success_rate = (self.stats['tasks']['total_completed'] / max(total_tasks_processed, 1)) * 100
|
||
|
||
return {
|
||
'summary': {
|
||
'uptime_seconds': uptime,
|
||
'uptime_formatted': self._format_uptime(uptime),
|
||
'requests_total': self.stats['requests']['total'],
|
||
'requests_per_second': round(requests_per_second, 2),
|
||
'error_rate_percent': round(error_rate, 2),
|
||
'active_tasks': len([t for t in task_status.values()
|
||
if t.get('status') in [TASK_STATUS_PENDING, TASK_STATUS_PROCESSING]])
|
||
},
|
||
'requests': {
|
||
'by_method': self.stats['requests']['by_method'],
|
||
'by_status': self.stats['requests']['by_status'],
|
||
'top_endpoints': dict(sorted(self.stats['requests']['by_endpoint'].items(),
|
||
key=lambda x: x[1], reverse=True)[:10])
|
||
},
|
||
'tasks': {
|
||
'total_created': self.stats['tasks']['total_created'],
|
||
'total_completed': self.stats['tasks']['total_completed'],
|
||
'total_failed': self.stats['tasks']['total_failed'],
|
||
'success_rate_percent': round(task_success_rate, 2),
|
||
'by_status': self.stats['tasks']['by_status']
|
||
},
|
||
'performance': {
|
||
'avg_response_time_ms': round(self.stats['performance']['avg_response_time'] * 1000, 2),
|
||
'max_response_time_ms': round(self.stats['performance']['max_response_time'] * 1000, 2),
|
||
'min_response_time_ms': round(self.stats['performance']['min_response_time'] * 1000, 2) if self.stats['performance']['min_response_time'] != float('inf') else 0
|
||
}
|
||
}
|
||
|
||
def _format_uptime(self, seconds):
|
||
"""Format uptime in human readable format."""
|
||
days, remainder = divmod(int(seconds), 86400)
|
||
hours, remainder = divmod(remainder, 3600)
|
||
minutes, seconds = divmod(remainder, 60)
|
||
|
||
parts = []
|
||
if days > 0:
|
||
parts.append(f"{days}d")
|
||
if hours > 0:
|
||
parts.append(f"{hours}h")
|
||
if minutes > 0:
|
||
parts.append(f"{minutes}m")
|
||
parts.append(f"{seconds}s")
|
||
|
||
return " ".join(parts)
|
||
|
||
|
||
# Global statistics collector
|
||
stats_collector = APIStatsCollector()
|
||
|
||
# get_task_status moved to shared.py
|
||
|
||
# cleanup_old_tasks moved to shared.py
|
||
|
||
def process_data_async(task_id, data_path, config_path, job_output_dir):
|
||
"""Background task to process data asynchronously."""
|
||
# 确保后台线程里有 Flask 应用上下文
|
||
with app.app_context():
|
||
logger.info(f"Job {task_id}: Background processing started for task {task_id}")
|
||
start_time = time.time()
|
||
|
||
try:
|
||
update_task_status(task_id, TASK_STATUS_PROCESSING, "开始处理数据...")
|
||
|
||
# 1. Load and override config FIRST
|
||
logger.info(f"Job {task_id}: Loading configuration from {config_path}")
|
||
config_start = time.time()
|
||
|
||
try:
|
||
with open(config_path, 'r') as f:
|
||
config = yaml.safe_load(f)
|
||
logger.info(f"Job {task_id}: Configuration loaded successfully with {len(config)} keys")
|
||
except Exception as e:
|
||
logger.error(f"Job {task_id}: Failed to load config from {config_path}: {str(e)}")
|
||
raise
|
||
|
||
# Update directories based on config output_dir
|
||
Config.update_directories_from_config(config_path)
|
||
|
||
# Sync app.config with updated directories
|
||
app.config['UPLOAD_FOLDER'] = Config.UPLOAD_FOLDER
|
||
app.config['OUTPUT_FOLDER'] = Config.OUTPUT_FOLDER
|
||
|
||
# Task status persistence now uses SQLite only
|
||
# JSON persistence has been disabled
|
||
# from .shared import set_task_status_file_path, load_task_status_from_file
|
||
# set_task_status_file_path(Config.OUTPUT_FOLDER / "task_status.json")
|
||
# load_task_status_from_file()
|
||
|
||
# Update job directories to be under the correct config-based paths
|
||
from pathlib import Path
|
||
job_upload_dir = Path(Config.UPLOAD_FOLDER) / task_id
|
||
job_output_dir = Path(Config.OUTPUT_FOLDER) / task_id
|
||
job_upload_dir.mkdir(parents=True, exist_ok=True)
|
||
job_output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Trigger an update to save output_dir to database
|
||
update_task_status(task_id, TASK_STATUS_PROCESSING, "目录已就绪", output_dir=str(job_output_dir))
|
||
|
||
# Move uploaded files to the correct config-based directories
|
||
try:
|
||
import shutil
|
||
|
||
# Move data file to correct uploads directory
|
||
if data_path.parent != job_upload_dir:
|
||
new_data_path = job_upload_dir / data_path.name
|
||
if data_path != new_data_path:
|
||
shutil.move(str(data_path), str(new_data_path))
|
||
data_path = new_data_path
|
||
logger.info(f"Job {task_id}: Moved data file to {data_path}")
|
||
|
||
# Move config file to correct uploads directory (if it's a custom config)
|
||
if config_path.parent != job_upload_dir and config_path.parent != Config.BASE_DIR:
|
||
new_config_path = job_upload_dir / config_path.name
|
||
if config_path != new_config_path:
|
||
shutil.move(str(config_path), str(new_config_path))
|
||
config_path = new_config_path
|
||
logger.info(f"Job {task_id}: Moved config file to {config_path}")
|
||
|
||
except Exception as e:
|
||
logger.warning(f"Job {task_id}: Failed to move uploaded files to configured directories: {str(e)}")
|
||
|
||
logger.debug(f"Job {task_id}: Using INI configured output directory: {Config.OUTPUT_FOLDER}")
|
||
logger.debug(f"Job {task_id}: Updated directories - Upload: {Config.UPLOAD_FOLDER}, Output: {Config.OUTPUT_FOLDER}, Job output: {job_output_dir}")
|
||
|
||
config_duration = time.time() - config_start
|
||
logger.info(f"Job {task_id}: Configuration processing completed in {config_duration:.3f}s")
|
||
|
||
update_task_status(task_id, TASK_STATUS_PROCESSING, "配置已加载，开始预处理...")
|
||
|
||
# 2. Data Preprocessing (files are already in correct directories)
|
||
logger.info(f"Job {task_id}: Starting preprocessing phase...")
|
||
preprocess_start = time.time()
|
||
|
||
processed_csv = data_path.parent / f"{data_path.stem}.processed.csv"
|
||
logger.debug(f"Job {task_id}: Input file: {data_path}, Output file: {processed_csv}")
|
||
|
||
process_file(str(data_path), str(processed_csv), str(config_path))
|
||
|
||
preprocess_duration = time.time() - preprocess_start
|
||
logger.info(f"Job {task_id}: Preprocessing completed in {preprocess_duration:.3f}s")
|
||
|
||
update_task_status(task_id, TASK_STATUS_PROCESSING, "预处理完成，开始GasFlux分析...")
|
||
|
||
# Write modified config to a temp file
|
||
final_config_path = data_path.parent / "final_config.yaml"
|
||
try:
|
||
with open(final_config_path, 'w') as f:
|
||
yaml.safe_dump(config, f)
|
||
logger.info(f"Job {task_id}: Final config written to {final_config_path}")
|
||
except Exception as e:
|
||
logger.error(f"Job {task_id}: Failed to write final config: {str(e)}")
|
||
raise
|
||
|
||
config_duration = time.time() - config_start
|
||
logger.info(f"Job {task_id}: Configuration processing completed in {config_duration:.3f}s")
|
||
|
||
update_task_status(task_id, TASK_STATUS_PROCESSING, "配置已加载，开始GasFlux分析...")
|
||
|
||
# 3. GasFlux Processing
|
||
logger.info(f"Job {task_id}: Starting GasFlux analysis...")
|
||
analysis_start = time.time()
|
||
|
||
processor = process_main(processed_csv, final_config_path, job_output_dir, task_id) # 获取返回值
|
||
|
||
analysis_duration = time.time() - analysis_start
|
||
logger.info(f"Job {task_id}: GasFlux analysis completed in {analysis_duration:.3f}s")
|
||
|
||
# 提取krig_params数据（只保存关键数值）
|
||
krig_params_data = []
|
||
if hasattr(processor, 'output_vars') and 'krig_parameters' in processor.output_vars:
|
||
for gas, params in processor.output_vars['krig_parameters'].items():
|
||
# 只保存数值类型的数据，跳过数组
|
||
clean_params = {}
|
||
for key, value in params.items():
|
||
if isinstance(value, (int, float)):
|
||
clean_params[key] = value
|
||
elif hasattr(value, 'item') and hasattr(value, 'size'):
|
||
# numpy数组：只处理单元素数组
|
||
if value.size == 1:
|
||
clean_params[key] = value.item()
|
||
# 多元素数组跳过，不保存
|
||
elif hasattr(value, 'item'):
|
||
# 其他numpy对象尝试转换
|
||
try:
|
||
clean_params[key] = value.item()
|
||
except ValueError:
|
||
# 转换失败则跳过
|
||
continue
|
||
|
||
krig_params_data.append({
|
||
'gas': gas,
|
||
'krig_params': clean_params
|
||
})
|
||
|
||
update_task_status(task_id, TASK_STATUS_PROCESSING, "GasFlux分析完成，正在生成报告...")
|
||
|
||
# Collect results and generate full URLs
|
||
logger.info(f"Job {task_id}: Collecting generated files from {job_output_dir}")
|
||
results_start = time.time()
|
||
results = []
|
||
|
||
# 先添加krig_params数据
|
||
results.extend(krig_params_data)
|
||
|
||
try:
|
||
for f in job_output_dir.rglob("*"):
|
||
if f.is_file():
|
||
rel_path = f.relative_to(app.config['OUTPUT_FOLDER']).as_posix()
|
||
file_size = f.stat().st_size
|
||
results.append({
|
||
"name": f.name,
|
||
"rel_path": rel_path,
|
||
"download_url": f"/download/{rel_path}", # Relative URL that client can use
|
||
"size": file_size
|
||
})
|
||
logger.debug(f"Job {task_id}: Found output file: {f.name} ({file_size} bytes)")
|
||
|
||
results_duration = time.time() - results_start
|
||
logger.info(f"Job {task_id}: Results collection completed in {results_duration:.3f}s - {len(results)} files generated")
|
||
|
||
total_size = sum(r.get('size', 0) for r in results)
|
||
logger.info(f"Job {task_id}: Total output size: {total_size} bytes across {len(results)} files")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Job {task_id}: Failed to collect results: {str(e)}")
|
||
raise
|
||
|
||
total_duration = time.time() - start_time
|
||
logger.info(f"Job {task_id}: Processing complete. Total duration: {total_duration:.3f}s, {len(results)} files generated.")
|
||
|
||
# Record task completion time for statistics
|
||
stats_collector.record_task_completion_time(total_duration)
|
||
|
||
update_task_status(task_id, TASK_STATUS_COMPLETED, "处理成功完成", results=results)
|
||
|
||
except Exception as e:
|
||
total_duration = time.time() - start_time
|
||
logger.error(f"Job {task_id}: Processing failed after {total_duration:.3f}s - Error: {str(e)}", exc_info=True)
|
||
|
||
# Record failed task processing time for statistics
|
||
stats_collector.record_task_completion_time(total_duration)
|
||
logger.error(f"Job {task_id}: Failed task details - Data: {data_path}, Config: {config_path}, Output: {job_output_dir}")
|
||
|
||
# Try to capture any partial results
|
||
partial_results = []
|
||
try:
|
||
for f in job_output_dir.rglob("*"):
|
||
if f.is_file():
|
||
rel_path = f.relative_to(app.config['OUTPUT_FOLDER']).as_posix()
|
||
partial_results.append({
|
||
"name": f.name,
|
||
"rel_path": rel_path,
|
||
"download_url": f"/download/{rel_path}", # Relative URL that client can use
|
||
"size": f.stat().st_size,
|
||
"note": "partial_result"
|
||
})
|
||
except Exception as collect_error:
|
||
logger.warning(f"Job {task_id}: Failed to collect partial results: {str(collect_error)}")
|
||
|
||
error_msg = f"处理失败: {str(e)}"
|
||
if partial_results:
|
||
error_msg += f" (部分结果可用: {len(partial_results)} 个文件)"
|
||
|
||
update_task_status(task_id, TASK_STATUS_FAILED, error=error_msg, results=partial_results if partial_results else None)
|
||
|
||
# Import GasFlux modules
|
||
logger.info("Importing GasFlux modules...")
|
||
import_start = time.time()
|
||
|
||
try:
|
||
# Try absolute imports first (more reliable)
|
||
from src.gasflux.processing_pipelines import process_main
|
||
from src.gasflux.data_processor import process_file
|
||
from src.gasflux.reporting import generate_reports
|
||
import_duration = time.time() - import_start
|
||
logger.info(f"GasFlux modules imported successfully in {import_duration:.3f}s (absolute import)")
|
||
except ImportError as e1:
|
||
logger.warning(f"Absolute import failed, trying relative import: {e1}")
|
||
try:
|
||
from .processing_pipelines import process_main
|
||
from .data_processor import process_file
|
||
from .reporting import generate_reports
|
||
import_duration = time.time() - import_start
|
||
logger.info(f"GasFlux modules imported successfully in {import_duration:.3f}s (relative import)")
|
||
except ImportError as e2:
|
||
import_duration = time.time() - import_start
|
||
logger.error(f"Failed to import GasFlux modules after {import_duration:.3f}s - Absolute error: {e1}, Relative error: {e2}")
|
||
raise ImportError(f"Cannot import GasFlux modules: {e2}")
|
||
|
||
app = Flask(__name__)
|
||
CORS(app) # Initialize CORS
|
||
|
||
# Enhanced logging configuration after app initialization
|
||
try:
|
||
log_file_path = Path(Config.LOG_FILE)
|
||
log_file_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Create file handler
|
||
file_handler = logging.FileHandler(log_file_path, encoding='utf-8')
|
||
file_handler.setLevel(Config.get_log_level())
|
||
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||
file_handler.setFormatter(formatter)
|
||
|
||
# Add file handler to logger
|
||
logger.addHandler(file_handler)
|
||
logger.info(f"File logging initialized. Log file: {log_file_path.absolute()}")
|
||
print(f"Log file: {log_file_path.absolute()}") # Also print to console
|
||
except Exception as e:
|
||
print(f"Warning: Failed to initialize file logging: {e}")
|
||
logger.warning(f"Failed to initialize file logging: {e}")
|
||
|
||
logger.info("Flask application initialized")
|
||
|
||
# Request logging middleware
|
||
@app.before_request
|
||
def log_request_info():
|
||
"""Log incoming request details."""
|
||
g.start_time = time.time()
|
||
logger.info(f"REQUEST: {request.method} {request.url} - IP: {request.remote_addr} - User-Agent: {request.headers.get('User-Agent', 'Unknown')}")
|
||
|
||
@app.after_request
|
||
def log_response_info(response):
|
||
"""Log response details."""
|
||
duration = time.time() - g.start_time
|
||
logger.info(f"RESPONSE: {request.method} {request.url} - Status: {response.status_code} - Duration: {duration:.3f}s")
|
||
|
||
# Record statistics
|
||
endpoint = request.url_rule.rule if request.url_rule else request.path
|
||
stats_collector.record_request(request.method, endpoint, response.status_code, duration)
|
||
|
||
return response
|
||
|
||
# Initialize configuration from environment variables
|
||
Config.init_base_dir()
|
||
|
||
# Apply configuration to app (directories will be created dynamically based on config)
|
||
# ALLOWED_DATA_EXTENSIONS and ALLOWED_CONFIG_EXTENSIONS moved to shared.py
|
||
|
||
app.config['MAX_CONTENT_LENGTH'] = Config.MAX_CONTENT_LENGTH
|
||
app.config['BASE_URL'] = Config.BASE_URL
|
||
|
||
# Set upload and output folders from config
|
||
app.config['UPLOAD_FOLDER'] = Config.UPLOAD_FOLDER
|
||
app.config['OUTPUT_FOLDER'] = Config.OUTPUT_FOLDER
|
||
|
||
# Database and persistence configuration
|
||
if Config.DB_PATH:
|
||
app.config['DB_PATH'] = Config.DB_PATH
|
||
if Config.TASK_PERSIST_BACKEND:
|
||
app.config['TASK_PERSIST_BACKEND'] = Config.TASK_PERSIST_BACKEND
|
||
app.config['JANITOR_DRY_RUN'] = str(Config.JANITOR_DRY_RUN).lower()
|
||
if Config.ADMIN_BOOTSTRAP_KEY:
|
||
app.config['ADMIN_BOOTSTRAP_KEY'] = Config.ADMIN_BOOTSTRAP_KEY
|
||
|
||
# Task cleanup configuration
|
||
app.config['SUCCESSFUL_TASK_CLEANUP_AGE'] = Config.SUCCESSFUL_TASK_CLEANUP_AGE
|
||
app.config['FAILED_TASK_CLEANUP_AGE'] = Config.FAILED_TASK_CLEANUP_AGE
|
||
app.config['TASK_CLEANUP_INTERVAL'] = Config.TASK_CLEANUP_INTERVAL
|
||
|
||
# Debug logging for cleanup configuration
|
||
logger.info(f"App config: FAILED_TASK_CLEANUP_AGE = {app.config['FAILED_TASK_CLEANUP_AGE']}")
|
||
logger.info(f"App config: TASK_CLEANUP_INTERVAL = {app.config['TASK_CLEANUP_INTERVAL']}")
|
||
|
||
# Log current configuration
|
||
logger.info(f"Upload folder: {Config.UPLOAD_FOLDER}")
|
||
logger.info(f"Output folder: {Config.OUTPUT_FOLDER}")
|
||
logger.info(f"Configuration: {Config.to_dict()}")
|
||
|
||
# Ensure directories exist at startup
|
||
def setup_directories():
|
||
logger.info("Initializing application directories...")
|
||
start_time = time.time()
|
||
|
||
try:
|
||
# Check if directories already exist
|
||
upload_exists = Config.UPLOAD_FOLDER.exists()
|
||
output_exists = Config.OUTPUT_FOLDER.exists()
|
||
|
||
Config.UPLOAD_FOLDER.mkdir(parents=True, exist_ok=True)
|
||
Config.OUTPUT_FOLDER.mkdir(parents=True, exist_ok=True)
|
||
|
||
duration = time.time() - start_time
|
||
logger.info(f"Directories initialized in {duration:.3f}s: {Config.UPLOAD_FOLDER} ({'existing' if upload_exists else 'created'}), {Config.OUTPUT_FOLDER} ({'existing' if output_exists else 'created'})")
|
||
|
||
# Log directory permissions
|
||
upload_writable = os.access(Config.UPLOAD_FOLDER, os.W_OK)
|
||
output_writable = os.access(Config.OUTPUT_FOLDER, os.W_OK)
|
||
logger.info(f"Directory permissions - Upload writable: {upload_writable}, Output writable: {output_writable}")
|
||
|
||
except Exception as e:
|
||
duration = time.time() - start_time
|
||
logger.error(f"Failed to create directories after {duration:.3f}s: {e}")
|
||
raise
|
||
|
||
# setup_directories() - commented out to avoid creating directories at startup
|
||
# Directories will be created dynamically based on config when processing tasks
|
||
|
||
# allowed_file moved to shared.py
|
||
|
||
# Initialize database and start background services
|
||
from .db import init_app as init_db
|
||
init_db(app)
|
||
|
||
# Import blueprints after app initialization to avoid circular imports
|
||
from .blueprints.health import health_bp
|
||
from .blueprints.upload import upload_bp
|
||
from .blueprints.tasks import tasks_bp
|
||
from .blueprints.task_pool import task_pool_bp
|
||
from .blueprints.stats import stats_bp
|
||
from .blueprints.config import config_bp
|
||
from .blueprints.reports import reports_bp
|
||
from .blueprints.download import download_bp
|
||
from .blueprints.web import web_bp
|
||
from .blueprints.api_keys import api_keys_bp
|
||
|
||
# Register blueprints
|
||
app.register_blueprint(health_bp)
|
||
app.register_blueprint(upload_bp)
|
||
app.register_blueprint(tasks_bp)
|
||
app.register_blueprint(task_pool_bp)
|
||
app.register_blueprint(stats_bp)
|
||
app.register_blueprint(config_bp)
|
||
app.register_blueprint(reports_bp)
|
||
app.register_blueprint(download_bp)
|
||
app.register_blueprint(web_bp)
|
||
app.register_blueprint(api_keys_bp)
|
||
|
||
# Task status persistence now uses SQLite only
|
||
# JSON persistence has been disabled - functions removed from shared.py
|
||
|
||
# Initialize janitor for background cleanup
|
||
try:
|
||
from .janitor import start_janitor, reconcile_tasks_on_startup
|
||
with app.app_context():
|
||
reconcile_tasks_on_startup()
|
||
# No longer need to load task status into memory
|
||
start_janitor(app)
|
||
except Exception as e:
|
||
print(f"⚠ Failed to setup task persistence: {e}")
|
||
|
||
# _get_file_type and _format_response moved to shared.py
|
||
|
||
|
||
if __name__ == '__main__':
|
||
app.run(host=Config.HOST, port=Config.PORT, debug=Config.DEBUG)
|