增加web_api

This commit is contained in:
2026-02-05 15:13:54 +08:00
parent 443ec09c5c
commit d5edbc0723
43 changed files with 7036 additions and 2640 deletions

728
src/gasflux/app.py Normal file
View File

@ -0,0 +1,728 @@
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, TASK_STATUS_PENDING, TASK_STATUS_PROCESSING, TASK_STATUS_COMPLETED, TASK_STATUS_FAILED
except ImportError:
# Fallback to absolute import (when run directly)
from shared import task_status, TASK_STATUS_PENDING, TASK_STATUS_PROCESSING, TASK_STATUS_COMPLETED, TASK_STATUS_FAILED
# Blueprints will be imported after app initialization to avoid circular imports
# Environment-based configuration management
class Config:
"""Configuration management using environment variables with defaults."""
# Server configuration
HOST = os.getenv('GASFLUX_HOST', '0.0.0.0')
PORT = int(os.getenv('GASFLUX_PORT', '5000'))
DEBUG = os.getenv('GASFLUX_DEBUG', 'false').lower() in ('true', '1', 'yes', 'on')
# Directory configuration
BASE_DIR = None # Will be set dynamically
UPLOAD_FOLDER_NAME = os.getenv('GASFLUX_UPLOAD_FOLDER', 'web_api_data/uploads')
OUTPUT_FOLDER_NAME = os.getenv('GASFLUX_OUTPUT_FOLDER', 'web_api_data/outputs')
# File size limits (in bytes)
MAX_CONTENT_LENGTH = int(os.getenv('GASFLUX_MAX_CONTENT_LENGTH', str(100 * 1024 * 1024))) # 100MB
# Logging configuration
LOG_LEVEL = os.getenv('GASFLUX_LOG_LEVEL', 'INFO').upper()
LOG_FILE = os.getenv('GASFLUX_LOG_FILE', 'logs/gasflux_api.log')
# CORS configuration
CORS_ORIGINS = os.getenv('GASFLUX_CORS_ORIGINS', '*').split(',')
# Task management
TASK_CLEANUP_INTERVAL = int(os.getenv('GASFLUX_TASK_CLEANUP_INTERVAL', '3600')) # 1 hour in seconds
MAX_TASK_AGE = int(os.getenv('GASFLUX_MAX_TASK_AGE', str(24 * 3600))) # 24 hours in seconds
# Performance tuning
THREADS = int(os.getenv('GASFLUX_THREADS', '8')) # Waitress threads
CONNECTION_LIMIT = int(os.getenv('GASFLUX_CONNECTION_LIMIT', '100'))
CHANNEL_TIMEOUT = int(os.getenv('GASFLUX_CHANNEL_TIMEOUT', '300')) # 5 minutes
@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."""
if output_dir:
# Use config-based output directory
output_base = Path(output_dir)
if not output_base.is_absolute():
output_base = cls.BASE_DIR / output_base
else:
# Use default relative paths
output_base = cls.BASE_DIR
# Set upload and output directories relative to output_base
cls.UPLOAD_FOLDER = output_base / "uploads"
cls.OUTPUT_FOLDER = output_base / "outputs"
# 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}")
@classmethod
def update_directories_from_config(cls, config_path=None):
"""Update directories based on config file."""
if config_path and Path(config_path).exists():
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
output_dir = config.get('output_dir')
if output_dir:
cls.init_directories(output_dir)
logger.info(f"Directories updated from config: {config_path}")
except Exception as e:
logger.warning(f"Failed to update directories from config {config_path}: {e}")
else:
logger.info("Using default directory configuration")
@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,
'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):
"""Update task status in the global dictionary."""
timestamp = time.time()
old_status = task_status.get(task_id, {}).get("status", "unknown")
task_status[task_id] = {
"status": status,
"message": message,
"results": results,
"error": error,
"updated_at": timestamp
}
# Log detailed status change with context
log_msg = f"Task {task_id} status changed: {old_status} -> {status}"
if message:
log_msg += f" | Message: {message}"
if results:
log_msg += f" | Results count: {len(results) if isinstance(results, list) else 'N/A'}"
if error:
log_msg += f" | Error: {error}"
log_level = logging.ERROR if status == TASK_STATUS_FAILED else logging.INFO
logger.log(log_level, log_msg)
# Update task statistics
stats_collector.record_task_status_change(old_status, status)
# 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."""
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, "Starting data 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
# Update task status file path to new output directory
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)
# 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}: Keeping original output directory: {config.get('output_dir', 'not set')}")
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, "Configuration loaded, starting preprocessing...")
# 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, "Preprocessing completed, starting GasFlux analysis...")
# 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, "Configuration loaded, starting GasFlux analysis...")
# 3. GasFlux Processing
logger.info(f"Job {task_id}: Starting GasFlux analysis...")
analysis_start = time.time()
process_main(processed_csv, final_config_path, task_id)
analysis_duration = time.time() - analysis_start
logger.info(f"Job {task_id}: GasFlux analysis completed in {analysis_duration:.3f}s")
update_task_status(task_id, TASK_STATUS_PROCESSING, "GasFlux analysis completed, generating reports...")
# Collect results and generate full URLs
logger.info(f"Job {task_id}: Collecting generated files from {job_output_dir}")
results_start = time.time()
results = []
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, "Processing completed successfully", 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"Processing failed: {str(e)}"
if partial_results:
error_msg += f" (partial results available: {len(partial_results)} files)"
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
# Don't set UPLOAD_FOLDER and OUTPUT_FOLDER here - they will be set dynamically per request
# Set defaults to avoid KeyError if any handler reads before config is applied
app.config.setdefault('UPLOAD_FOLDER', None)
app.config.setdefault('OUTPUT_FOLDER', None)
# 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
# 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
# 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)
# Load persisted task status after app initialization
try:
from .shared import (
load_task_status_from_file,
save_task_status_to_file,
set_task_status_file_path,
)
# 只有在 OUTPUT_FOLDER 有效时才启用持久化
if hasattr(Config, 'OUTPUT_FOLDER') and Config.OUTPUT_FOLDER:
task_status_path = Config.OUTPUT_FOLDER / "task_status.json"
set_task_status_file_path(task_status_path)
logger.info(f"Task status persistence path set to: {task_status_path}")
with app.app_context():
load_task_status_from_file()
import atexit
def _save_on_exit():
with app.app_context():
save_task_status_to_file()
atexit.register(_save_on_exit)
else:
logger.info("Task status persistence will be configured after config is loaded")
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='0.0.0.0', port=5000, debug=False)

View File

@ -0,0 +1 @@
# GasFlux API Blueprints

View File

@ -0,0 +1,67 @@
"""
Configuration Blueprint
Provides configuration information and environment variables endpoints.
"""
import os
from flask import Blueprint
from ..app import Config
from ..shared import _format_response, log_performance, logger
# Create blueprint
config_bp = Blueprint('config', __name__, url_prefix='/config')
@config_bp.route('', methods=['GET'])
@log_performance
def get_config():
"""Get current configuration (without sensitive information)."""
logger.debug("Configuration requested")
try:
config_info = Config.to_dict()
# Remove potentially sensitive information
safe_config = config_info.copy()
data = {
"configuration": safe_config,
"environment_variables": {
"supported": [
"GASFLUX_HOST",
"GASFLUX_PORT",
"GASFLUX_DEBUG",
"GASFLUX_UPLOAD_FOLDER",
"GASFLUX_OUTPUT_FOLDER",
"GASFLUX_MAX_CONTENT_LENGTH",
"GASFLUX_LOG_LEVEL",
"GASFLUX_LOG_FILE",
"GASFLUX_CORS_ORIGINS",
"GASFLUX_TASK_CLEANUP_INTERVAL",
"GASFLUX_MAX_TASK_AGE",
"GASFLUX_THREADS",
"GASFLUX_CONNECTION_LIMIT",
"GASFLUX_CHANNEL_TIMEOUT"
],
"current_values": {
key: os.getenv(key, "not set") if key.startswith("GASFLUX_") else "internal"
for key in [
"GASFLUX_HOST", "GASFLUX_PORT", "GASFLUX_DEBUG",
"GASFLUX_UPLOAD_FOLDER", "GASFLUX_OUTPUT_FOLDER",
"GASFLUX_MAX_CONTENT_LENGTH", "GASFLUX_LOG_LEVEL",
"GASFLUX_LOG_FILE", "GASFLUX_CORS_ORIGINS",
"GASFLUX_TASK_CLEANUP_INTERVAL", "GASFLUX_MAX_TASK_AGE",
"GASFLUX_THREADS", "GASFLUX_CONNECTION_LIMIT",
"GASFLUX_CHANNEL_TIMEOUT"
]
}
}
}
return _format_response(200, "配置信息获取成功", data)
except Exception as e:
logger.error(f"Failed to retrieve configuration: {str(e)}", exc_info=True)
return _format_response(500, "获取配置信息失败", {
"error_details": str(e)
})

View File

@ -0,0 +1,68 @@
"""
Download Blueprint
Handles file download endpoints.
"""
from pathlib import Path
from flask import Blueprint, send_file, current_app
from ..shared import _format_response, log_performance, logger
# Create blueprint
download_bp = Blueprint('download', __name__, url_prefix='/download')
@download_bp.route('/<path:filename>')
@log_performance
def download_file(filename):
"""Download a processed file."""
from flask import request
logger.info(f"Download request for file: {filename} from IP {request.remote_addr}")
try:
# 支持两种路径格式:
# 1. 绝对路径(以 / 开头,如 /full/path/to/file
# 2. 相对路径task_id/filename
if filename.startswith('/'):
# 绝对路径 - 直接使用
file_path = Path(filename)
else:
# 相对路径 - 相对于 OUTPUT_FOLDER
output_folder = Path(current_app.config.get('OUTPUT_FOLDER') or '')
if not output_folder:
logger.error("OUTPUT_FOLDER not configured")
return _format_response(500, "服务器配置错误")
# 解析 task_id/filename 格式
parts = filename.split('/', 1)
if len(parts) != 2:
logger.warning(f"Invalid relative path format: {filename}")
return _format_response(400, "无效的文件路径")
task_id, filename_part = parts
file_path = output_folder / task_id / filename_part
# Security check - ensure file is within output folder
file_path = file_path.resolve()
output_folder = Path(current_app.config.get('OUTPUT_FOLDER') or '').resolve()
if output_folder and not str(file_path).startswith(str(output_folder)):
logger.warning(f"Security violation: Attempted to access file outside output folder: {filename}")
return _format_response(403, "访问被拒绝")
if not file_path.exists():
logger.warning(f"File not found: {filename}")
return _format_response(404, "文件未找到")
if not file_path.is_file():
logger.warning(f"Path is not a file: {filename}")
return _format_response(400, "不是文件")
file_size = file_path.stat().st_size
logger.info(f"Serving file: {filename} ({file_size} bytes)")
return send_file(file_path)
except Exception as e:
logger.error(f"Error serving file {filename}: {str(e)}", exc_info=True)
return _format_response(500, "内部服务器错误")

View File

@ -0,0 +1,94 @@
"""
Health Check Blueprint
Provides API health monitoring and system status endpoints.
"""
import os
import time
import logging
from flask import Blueprint, request
from ..app import Config, stats_collector
from ..shared import task_status, TASK_STATUS_PENDING, TASK_STATUS_PROCESSING
from ..shared import _format_response, log_performance, logger
# Create blueprint
health_bp = Blueprint('health', __name__, url_prefix='/health')
@health_bp.route('', methods=['GET'])
@log_performance
def health_check():
"""API Health Check"""
logger.debug("Health check requested")
try:
# Check storage accessibility
uploads_writable = os.access(Config.UPLOAD_FOLDER, os.W_OK)
outputs_writable = os.access(Config.OUTPUT_FOLDER, os.W_OK)
# Check active tasks
active_tasks = len([t for t in task_status.values() if t.get("status") in [TASK_STATUS_PENDING, TASK_STATUS_PROCESSING]])
# Get basic stats for health check
stats_summary = stats_collector.get_summary()
health_data = {
"status": "healthy",
"version": "1.0.0",
"timestamp": time.time(),
"uptime": stats_summary['summary']['uptime_formatted'],
"storage": {
"uploads_writable": uploads_writable,
"outputs_writable": outputs_writable
},
"tasks": {
"active_count": active_tasks,
"total_tracked": len(task_status),
"total_processed": stats_summary['tasks']['total_completed'] + stats_summary['tasks']['total_failed'],
"success_rate_percent": stats_summary['tasks']['success_rate_percent']
},
"performance": {
"requests_per_second": stats_summary['summary']['requests_per_second'],
"avg_response_time_ms": stats_summary['performance']['avg_response_time_ms'],
"error_rate_percent": stats_summary['summary']['error_rate_percent']
}
}
# Determine health status based on metrics
is_healthy = True
issues = []
if not uploads_writable:
issues.append("上传文件夹不可写")
is_healthy = False
if not outputs_writable:
issues.append("输出文件夹不可写")
is_healthy = False
if active_tasks > 20: # High load threshold
issues.append(f"活跃任务数量过多 ({active_tasks})")
if stats_summary['summary']['error_rate_percent'] > 10: # High error rate
issues.append(f"错误率过高 ({stats_summary['summary']['error_rate_percent']:.1f}%)")
is_healthy = False
health_data["status"] = "healthy" if is_healthy else "degraded"
if issues:
health_data["issues"] = issues
# Log warnings for potential issues
for issue in issues:
logger.warning(f"Health check issue: {issue}")
status_level = logging.DEBUG if is_healthy else logging.WARNING
logger.log(status_level, f"Health check: {health_data['status']} (active tasks: {active_tasks})")
status_code = 200 if is_healthy else 503 # 503 Service Unavailable for degraded
return _format_response(status_code, "健康检查完成" if is_healthy else "服务不可用", health_data)
except Exception as e:
logger.error(f"Health check failed: {str(e)}", exc_info=True)
return _format_response(500, "健康检查失败", {
"status": "unhealthy",
"error": str(e),
"timestamp": time.time()
})

View File

@ -0,0 +1,192 @@
"""
Reports Blueprint
Provides report listing and management endpoints.
"""
import time
from pathlib import Path
from flask import Blueprint, request, current_app
from ..shared import _get_file_type, _format_response, log_performance, logger, task_status
from ..app import Config
# Create blueprint
reports_bp = Blueprint('reports', __name__, url_prefix='/reports')
@reports_bp.route('', methods=['GET'])
@log_performance
def list_reports():
"""List all generated reports with pagination and filtering."""
logger.debug("Reports list requested")
try:
# Parse query parameters
try:
page = int(request.args.get('page', 1))
if page < 1:
return _format_response(400, "Invalid parameter: page must be >= 1")
except (ValueError, TypeError):
return _format_response(400, "Invalid parameter: page must be a valid integer")
try:
per_page = int(request.args.get('per_page', 20))
if per_page < 1 or per_page > 100:
return _format_response(400, "Invalid parameter: per_page must be between 1 and 100")
except (ValueError, TypeError):
return _format_response(400, "Invalid parameter: per_page must be a valid integer")
sort_by = request.args.get('sort_by', 'created_at')
sort_order = request.args.get('sort_order', 'desc')
status_filter = request.args.get('status', None) # 'completed', 'failed', or None for all
# Validate sort parameters
valid_sort_fields = ['created_at', 'task_id', 'file_size', 'processing_time']
if sort_by not in valid_sort_fields:
return _format_response(400, f"Invalid parameter: sort_by must be one of {valid_sort_fields}")
if sort_order not in ['asc', 'desc']:
return _format_response(400, "Invalid parameter: sort_order must be 'asc' or 'desc'")
# Validate status filter
valid_statuses = ['completed', 'failed', None]
if status_filter is not None and status_filter not in ['completed', 'failed']:
return _format_response(400, "Invalid parameter: status must be 'completed', 'failed', or not specified")
# 兼容缺省:优先 app.config其次 Config.OUTPUT_FOLDER
output_root = current_app.config.get('OUTPUT_FOLDER') or getattr(Config, 'OUTPUT_FOLDER', None)
if not output_root:
return _format_response(200, "报告列表获取成功", {
'reports': [],
'pagination': {'page': page, 'per_page': per_page, 'total_reports': 0, 'total_pages': 0, 'has_next': False, 'has_prev': False},
'filters': {'sort_by': sort_by, 'sort_order': sort_order, 'status': status_filter}
})
output_folder = Path(output_root)
reports = []
# Scan all task directories
if output_folder.exists():
for task_dir in output_folder.iterdir():
if not task_dir.is_dir():
continue
task_id = task_dir.name
# Get task information from global task_status
task_info = task_status.get(task_id, {})
task_status_value = task_info.get('status')
# Log task status for debugging
logger.debug(f"Task {task_id}: status from memory={task_status_value}, info={task_info}")
# 直接扫描平铺文件
files = [p for p in task_dir.iterdir() if p.is_file()]
if not files:
# 按需应用状态过滤
if status_filter:
continue
reports.append({
'task_id': task_id,
'report_name': "N/A",
'status': 'failed',
'created_at': task_dir.stat().st_mtime,
'file_count': 0,
'total_size': 0,
'processing_time_seconds': None,
'main_report': None,
'all_files': [],
'run_directory': f'{task_id}'
})
continue
# 识别主报告与统计
total_size = sum(f.stat().st_size for f in files)
created_at = max(f.stat().st_mtime for f in files) if files else task_dir.stat().st_mtime
def file_entry(p):
return {
'name': p.name,
'size': p.stat().st_size,
'type': _get_file_type(p.name),
# 使用相对路径下载,清晰且安全
'download_url': f"/download/{task_id}/{p.name}"
}
all_files = [file_entry(f) for f in files]
# 优先 CO2_report其次任意 *_report_*.html
report_html = None
for f in files:
if f.name.endswith('_report_') and f.suffix == '.html':
report_html = file_entry(f)
break
if not report_html:
for f in files:
if f.name.endswith('.html'):
report_html = file_entry(f)
break
# 任务状态:若有报告或关键产物则视为 completed
has_outputs = any(f.name.startswith(('config_', 'output_vars_', 'processed_data_', 'CO2_report_')) for f in files)
task_status_value = 'completed' if has_outputs else 'unknown'
if status_filter and task_status_value != status_filter:
continue
# Create report entry
report_entry = {
'task_id': task_id,
'report_name': task_id,
'status': task_status_value,
'created_at': created_at,
'file_count': len(files),
'total_size': total_size,
'processing_time_seconds': None,
'main_report': report_html,
'all_files': all_files,
'run_directory': f'{task_id}'
}
reports.append(report_entry)
# Sort reports
reverse_order = sort_order == 'desc'
if sort_by == 'created_at':
reports.sort(key=lambda x: x['created_at'], reverse=reverse_order)
elif sort_by == 'task_id':
reports.sort(key=lambda x: x['task_id'], reverse=reverse_order)
elif sort_by == 'file_size':
reports.sort(key=lambda x: x['total_size'], reverse=reverse_order)
elif sort_by == 'processing_time':
reports.sort(key=lambda x: x['processing_time_seconds'] or 0, reverse=reverse_order)
# Paginate results
total_reports = len(reports)
start_idx = (page - 1) * per_page
end_idx = start_idx + per_page
paginated_reports = reports[start_idx:end_idx]
# Calculate pagination metadata
total_pages = (total_reports + per_page - 1) // per_page
response_data = {
'reports': paginated_reports,
'pagination': {
'page': page,
'per_page': per_page,
'total_reports': total_reports,
'total_pages': total_pages,
'has_next': page < total_pages,
'has_prev': page > 1
},
'filters': {
'sort_by': sort_by,
'sort_order': sort_order,
'status': status_filter
}
}
logger.info(f"Returning {len(paginated_reports)} reports (page {page}/{total_pages})")
return _format_response(200, "报告列表获取成功", response_data)
except Exception as e:
logger.error(f"Error listing reports: {str(e)}", exc_info=True)
return _format_response(500, "内部服务器错误")

View File

@ -0,0 +1,93 @@
"""
Statistics Blueprint
Provides API statistics and monitoring endpoints.
"""
import time
from flask import Blueprint, current_app
from ..shared import _format_response, log_performance, logger,stats_collector, task_status
# Create blueprint
stats_bp = Blueprint('stats', __name__, url_prefix='/stats')
@stats_bp.route('', methods=['GET'])
@log_performance
def get_stats():
"""Get detailed API statistics and monitoring data."""
logger.debug("Statistics requested")
try:
# Get detailed statistics
stats_data = stats_collector.get_summary()
# Add current system information
try:
import psutil
memory = psutil.virtual_memory()
disk = psutil.disk_usage(str(current_app.config['OUTPUT_FOLDER']))
stats_data['system'] = {
'memory_usage_percent': memory.percent,
'memory_used_gb': round(memory.used / (1024**3), 2),
'memory_total_gb': round(memory.total / (1024**3), 2),
'disk_usage_percent': disk.percent,
'disk_used_gb': round(disk.used / (1024**3), 2),
'disk_total_gb': round(disk.total / (1024**3), 2)
}
except ImportError:
# psutil not available
stats_data['system'] = {
'note': 'System metrics unavailable - install psutil for detailed monitoring'
}
except Exception as e:
logger.warning(f"Failed to collect system metrics: {e}")
stats_data['system'] = {'error': str(e)}
# Add recent task information
recent_tasks = []
current_time = time.time()
for task_id, task_info in list(task_status.items())[-20:]: # Last 20 tasks
age = current_time - task_info.get('updated_at', 0)
recent_tasks.append({
'task_id': task_id,
'status': task_info.get('status'),
'age_seconds': round(age, 1),
'message': task_info.get('message', '')[:100] # Truncate long messages
})
stats_data['recent_tasks'] = recent_tasks
return _format_response(200, "统计信息获取成功", stats_data)
except Exception as e:
logger.error(f"Failed to retrieve statistics: {str(e)}", exc_info=True)
return _format_response(500, "获取统计信息失败", {
"error_details": str(e)
})
@stats_bp.route('/reset', methods=['POST'])
@log_performance
def reset_stats():
"""Reset API statistics (admin function)."""
logger.warning("Statistics reset requested")
try:
# Reset statistics
stats_collector.reset_stats()
# Log the reset
logger.info("API statistics have been reset")
return _format_response(200, "统计信息重置成功", {
"timestamp": time.time()
})
except Exception as e:
logger.error(f"Failed to reset statistics: {str(e)}", exc_info=True)
return _format_response(500, "重置统计信息失败", {
"error_details": str(e)
})

View File

@ -0,0 +1,228 @@
"""
Task Pool Blueprint
Handles task pool management endpoints: listing tasks with pagination, pool statistics.
"""
from flask import Blueprint, request
from pathlib import Path
from ..shared import (
get_task_list,
get_task_pool_stats,
_format_response,
log_performance,
logger,
task_status,
TASK_STATUS_PENDING,
TASK_STATUS_PROCESSING,
TASK_STATUS_COMPLETED,
TASK_STATUS_FAILED
)
# Create blueprint
task_pool_bp = Blueprint('task_pool', __name__, url_prefix='/tasks')
def _build_simple_downloads_from_results(results: list[dict]) -> dict:
"""
Build direct download shortcuts for common files, based on task results.
This is intentionally minimal and frontend-friendly.
"""
downloads: dict = {}
def set_once(key: str, url: str):
if key not in downloads and url:
downloads[key] = url
for item in results or []:
if not isinstance(item, dict):
continue
rel_path = item.get('rel_path')
if not rel_path:
continue
name_l = (item.get('name') or '').lower()
url = f"/download/{rel_path}"
if name_l.endswith('.xlsx'):
set_once('data_xlsx', url)
elif name_l.endswith('.xls'):
set_once('data_xls', url)
elif name_l.endswith('ch4_report.html'):
set_once('report_ch4', url)
elif name_l.endswith('co2_report.html'):
set_once('report_co2', url)
elif name_l.endswith(('.yaml', '.yml')):
set_once('config', url)
elif name_l.endswith('.json') and 'output_vars' in name_l:
set_once('metadata', url)
elif name_l.endswith('.html'):
# fallback: any html report
set_once('report_html', url)
return downloads
def _lean_task_summary(task_summary: dict) -> dict:
"""Return a minimal task representation for frontend consumption."""
task_id = task_summary.get('task_id')
status = task_summary.get('status')
lean = {
'task_id': task_id,
'status': status,
'message': task_summary.get('message'),
'updated_at': task_summary.get('updated_at'),
}
if status == TASK_STATUS_COMPLETED and task_id:
full_task_info = task_status.get(task_id, {})
results = full_task_info.get('results', []) or []
downloads = _build_simple_downloads_from_results(results)
if downloads:
lean['downloads'] = downloads
return lean
@task_pool_bp.route('', methods=['GET'])
@log_performance
def list_tasks():
"""Get paginated list of tasks with optional filtering."""
logger.debug(f"Task list request from IP {request.remote_addr}")
try:
# Parse query parameters
status_filter = request.args.get('status')
if status_filter:
# Support comma-separated status values
status_filter = status_filter.split(',')
page = int(request.args.get('page', 1))
page_size = int(request.args.get('page_size', 20))
sort_by = request.args.get('sort_by', 'updated_at')
sort_order = request.args.get('sort_order', 'desc')
# Validate parameters
if page < 1:
return _format_response(400, "页码必须大于0")
if page_size < 1 or page_size > 100:
return _format_response(400, "每页数量必须在1-100之间")
valid_sort_fields = ['created_at', 'updated_at', 'status']
if sort_by not in valid_sort_fields:
return _format_response(400, f"排序字段必须是以下之一: {', '.join(valid_sort_fields)}")
if sort_order.lower() not in ['asc', 'desc']:
return _format_response(400, "排序顺序必须是 'asc''desc'")
# Get task list
result = get_task_list(
status_filter=status_filter,
page=page,
page_size=page_size,
sort_by=sort_by,
sort_order=sort_order,
cleanup=False
)
# Slim response: only task status + downloads (completed only)
result['tasks'] = [_lean_task_summary(t) for t in result.get('tasks', [])]
logger.debug(f"Returning {len(result['tasks'])} tasks (page {page} of {result['total_pages']})")
return _format_response(200, "任务列表查询成功", result)
except ValueError as e:
logger.warning(f"Invalid parameter in task list request: {str(e)}")
return _format_response(400, "参数格式错误")
except Exception as e:
logger.error(f"Error listing tasks: {str(e)}", exc_info=True)
return _format_response(500, "内部服务器错误")
@task_pool_bp.route('/stats', methods=['GET'])
@log_performance
def get_pool_stats():
"""Get task pool statistics."""
logger.debug(f"Task pool stats request from IP {request.remote_addr}")
try:
stats = get_task_pool_stats()
logger.debug(f"Pool stats: {stats['total_tasks']} total tasks, "
f"{stats['active_tasks']} active, {stats['queued_tasks']} queued")
return _format_response(200, "任务池统计信息查询成功", stats)
except Exception as e:
logger.error(f"Error getting pool stats: {str(e)}", exc_info=True)
return _format_response(500, "内部服务器错误")
@task_pool_bp.route('/active', methods=['GET'])
@log_performance
def get_active_tasks():
"""Get list of currently active (processing) tasks."""
logger.debug(f"Active tasks request from IP {request.remote_addr}")
try:
# Get all processing tasks, no pagination needed for active tasks
result = get_task_list(
status_filter=TASK_STATUS_PROCESSING,
page=1,
page_size=1000, # Large page size to get all active tasks
sort_by='updated_at',
sort_order='asc', # Oldest first
cleanup=False
)
active_tasks = result['tasks']
active_tasks = [_lean_task_summary(t) for t in active_tasks]
logger.debug(f"Returning {len(active_tasks)} active tasks")
return _format_response(200, "活跃任务查询成功", {
'active_tasks': active_tasks,
'count': len(active_tasks)
})
except Exception as e:
logger.error(f"Error getting active tasks: {str(e)}", exc_info=True)
return _format_response(500, "内部服务器错误")
@task_pool_bp.route('/queue', methods=['GET'])
@log_performance
def get_queued_tasks():
"""Get list of queued (pending) tasks."""
logger.debug(f"Queued tasks request from IP {request.remote_addr}")
try:
# Get all pending tasks, sorted by creation time
result = get_task_list(
status_filter=TASK_STATUS_PENDING,
page=1,
page_size=1000, # Large page size to get all queued tasks
sort_by='created_at',
sort_order='asc', # Oldest first (FIFO)
cleanup=False
)
queued_tasks = result['tasks']
queued_tasks = [_lean_task_summary(t) for t in queued_tasks]
logger.debug(f"Returning {len(queued_tasks)} queued tasks")
return _format_response(200, "队列任务查询成功", {
'queued_tasks': queued_tasks,
'count': len(queued_tasks),
'queue_position_info': "任务按创建时间排序,较早的任务优先处理"
})
except Exception as e:
logger.error(f"Error getting queued tasks: {str(e)}", exc_info=True)
return _format_response(500, "内部服务器错误")

View File

@ -0,0 +1,220 @@
"""
Tasks Blueprint
Handles task management endpoints: status query, update, and deletion.
"""
from flask import Blueprint, request
from ..shared import (
get_task_status,
update_task_status,
cleanup_old_tasks,
_format_response,
log_performance,
logger,
task_status,
_build_simple_downloads_from_results,
TASK_STATUS_COMPLETED,
TASK_STATUS_FAILED,
TASK_STATUS_PROCESSING,
TASK_STATUS_PENDING,
)
# Create blueprint
tasks_bp = Blueprint('tasks', __name__, url_prefix='/task')
@tasks_bp.route('/<task_id>', methods=['GET'])
@log_performance
def get_task_status_endpoint(task_id):
"""Get the status of a processing task."""
logger.debug(f"Status request for task {task_id}")
try:
# Note: cleanup_old_tasks() is disabled for individual task queries
# to preserve historical task data for task pool management
# cleanup_old_tasks()
task_info = get_task_status(task_id)
if task_info.get("status") == "not_found":
logger.warning(f"Status request for non-existent task {task_id} from IP {request.remote_addr}")
return _format_response(404, "任务未找到")
data = {
"task_id": task_id,
"status": task_info["status"],
"message": task_info.get("message", ""),
"updated_at": task_info.get("updated_at", 0)
}
if task_info["status"] == TASK_STATUS_COMPLETED:
results = task_info.get("results", [])
data["results"] = results
# Add direct download shortcuts for frontend (if available or can be derived)
downloads = task_info.get("downloads") or _build_simple_downloads_from_results(results)
if downloads:
data["downloads"] = downloads
logger.debug(f"Task {task_id}: Returning {len(results)} completed results")
return _format_response(200, "任务查询成功", data)
elif task_info["status"] == TASK_STATUS_FAILED:
error_msg = task_info.get("error", "未知错误")
data["error"] = error_msg
logger.warning(f"Task {task_id}: Returning failure status - {error_msg}")
# Return 200 for failed tasks since this is expected behavior, not an HTTP error
return _format_response(200, "任务处理失败", data)
else:
# Processing or pending status
return _format_response(200, "任务查询成功", data)
except Exception as e:
logger.error(f"Error retrieving status for task {task_id}: {str(e)}", exc_info=True)
return _format_response(500, "内部服务器错误")
@tasks_bp.route('/<task_id>', methods=['PUT'])
@log_performance
def update_task(task_id):
"""Update task status and information."""
logger.info(f"Task update request for {task_id} from IP {request.remote_addr}")
try:
# Validate task exists
task_info = get_task_status(task_id)
if task_info.get("status") == "not_found":
logger.warning(f"Update request for non-existent task {task_id}")
return _format_response(404, "任务未找到")
# Parse request data
data = request.get_json()
if not data:
return _format_response(400, "请求体必须是 JSON 格式")
# Validate allowed fields
allowed_fields = ['status', 'message', 'priority']
valid_statuses = [TASK_STATUS_PENDING, TASK_STATUS_PROCESSING,
TASK_STATUS_COMPLETED, TASK_STATUS_FAILED]
updates = {}
for field in allowed_fields:
if field in data:
if field == 'status' and data[field] not in valid_statuses:
return _format_response(400, f"无效状态。必须是以下之一: {', '.join(valid_statuses)}")
updates[field] = data[field]
if not updates:
return _format_response(400, "没有有效的字段可更新")
# Update task status
current_status = task_info.get('status')
new_status = updates.get('status', current_status)
message = updates.get('message', task_info.get('message'))
# Special handling for status changes
if 'status' in updates:
if new_status == TASK_STATUS_COMPLETED:
# For completed tasks, we might want to add fake results if none exist
if not task_info.get('results'):
logger.warning(f"Marking task {task_id} as completed but no results found")
elif new_status == TASK_STATUS_FAILED:
# For failed tasks, error message is required
error_msg = updates.get('message', 'Task manually marked as failed')
update_task_status(task_id, new_status, error_msg)
else:
update_task_status(task_id, new_status, message)
else:
# Only update message
update_task_status(task_id, current_status, message)
# Update priority if provided
if 'priority' in updates:
task_status[task_id]['priority'] = updates['priority']
# Get updated task info
updated_task = get_task_status(task_id)
data = {
"task_id": task_id,
"status": "updated",
"task_info": {
"status": updated_task.get("status"),
"message": updated_task.get("message"),
"updated_at": updated_task.get("updated_at", 0),
"priority": updated_task.get("priority", "normal")
}
}
logger.info(f"Task {task_id} updated: {updates}")
return _format_response(200, "任务更新成功", data)
except Exception as e:
logger.error(f"Error updating task {task_id}: {str(e)}", exc_info=True)
return _format_response(500, "内部服务器错误")
@tasks_bp.route('/<task_id>', methods=['DELETE'])
@log_performance
def delete_task(task_id):
"""Delete a task and its associated files."""
logger.info(f"Task deletion request for {task_id} from IP {request.remote_addr}")
try:
# Validate task exists
task_info = get_task_status(task_id)
if task_info.get("status") == "not_found":
logger.warning(f"Delete request for non-existent task {task_id}")
return _format_response(404, "任务未找到")
# Check if task is currently processing
if task_info.get("status") in [TASK_STATUS_PROCESSING, TASK_STATUS_PENDING]:
return _format_response(409, "无法删除当前正在处理或等待处理的任务", {
"task_status": task_info.get("status")
})
# Delete associated files
from pathlib import Path
from flask import current_app
import shutil
output_folder = Path(current_app.config['OUTPUT_FOLDER'])
task_folder = output_folder / task_id
files_deleted = 0
total_size_deleted = 0
if task_folder.exists():
try:
# Calculate total size before deletion
for file_path in task_folder.rglob('*'):
if file_path.is_file():
total_size_deleted += file_path.stat().st_size
# Delete the entire task folder
shutil.rmtree(task_folder)
files_deleted = 1 # Count as one folder deleted
logger.info(f"Deleted task folder: {task_folder}")
except Exception as e:
logger.error(f"Error deleting task folder {task_folder}: {str(e)}")
return _format_response(500, f"删除任务文件失败: {str(e)}")
# Remove from task status tracking
if task_id in task_status:
del task_status[task_id]
logger.info(f"Removed task {task_id} from status tracking")
data = {
"task_id": task_id,
"status": "deleted",
"details": {
"folders_deleted": files_deleted,
"total_size_deleted": total_size_deleted,
"task_status": task_info.get("status")
}
}
logger.info(f"Task {task_id} deleted successfully")
return _format_response(200, "任务及相关文件删除成功", data)
except Exception as e:
logger.error(f"Error deleting task {task_id}: {str(e)}", exc_info=True)
return _format_response(500, "内部服务器错误")

View File

@ -0,0 +1,134 @@
"""
Upload Blueprint
Handles file upload and processing initiation endpoints.
"""
import uuid
import threading
from pathlib import Path
from flask import Blueprint, request, current_app
from werkzeug.utils import secure_filename
import yaml
from io import BytesIO
from ..app import process_data_async
from ..shared import _format_response, log_performance, logger, ALLOWED_DATA_EXTENSIONS, ALLOWED_CONFIG_EXTENSIONS, allowed_file,update_task_status, TASK_STATUS_PENDING, TASK_STATUS_FAILED
# Create blueprint
upload_bp = Blueprint('upload', __name__, url_prefix='/upload')
@upload_bp.route('', methods=['POST'])
@log_performance
def upload_file():
logger.info("Received upload request")
logger.info(f"Request content length: {request.content_length} bytes")
# Check if data file is present
if 'file' not in request.files:
logger.warning("Upload failed: No data file part in request")
return _format_response(400, "未找到数据文件部分")
data_file = request.files['file']
config_file = request.files.get('config')
# Log file details
logger.info(f"Data file: {data_file.filename} (size: {getattr(data_file, 'content_length', 'unknown')} bytes)")
if config_file:
logger.info(f"Config file: {config_file.filename} (size: {getattr(config_file, 'content_length', 'unknown')} bytes)")
else:
logger.info("No custom config file provided, will use default")
if data_file.filename == '':
logger.warning("Upload failed: No data file selected (empty filename)")
return _format_response(400, "未选择数据文件")
if not allowed_file(data_file.filename, ALLOWED_DATA_EXTENSIONS):
logger.warning(f"Upload failed: Invalid data file type {data_file.filename} - allowed: {ALLOWED_DATA_EXTENSIONS}")
return _format_response(400, "无效的数据文件类型。只允许 .xlsx 和 .xls 格式。")
# Generate unique job ID
job_id = str(uuid.uuid4())
logger.info(f"Generated job ID: {job_id}")
# 1) Parse config content (parse in memory without saving first)
if config_file and config_file.filename != '':
if not allowed_file(config_file.filename, ALLOWED_CONFIG_EXTENSIONS):
return _format_response(400, "无效的配置文件类型。只允许 .yaml 和 .yml 格式。")
config_file.stream.seek(0)
config_text = config_file.read().decode('utf-8', errors='ignore')
try:
active_config = yaml.safe_load(config_text)
except Exception:
return _format_response(400, "配置文件解析失败")
# Reset stream for saving
config_file.stream = BytesIO(config_text.encode('utf-8'))
else:
default_config_path = Path(__file__).parent.parent / "gasflux_config.yaml"
with open(default_config_path, 'r', encoding='utf-8') as f:
active_config = yaml.safe_load(f)
# 2) Create job directories based on config['output_dir']
output_base = Path(active_config['output_dir']).expanduser()
job_upload_dir = output_base / "uploads" / job_id
job_output_dir = output_base / "outputs" / job_id
job_upload_dir.mkdir(parents=True, exist_ok=True)
job_output_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"Job {job_id}: Created directories - Upload: {job_upload_dir}, Output: {job_output_dir}")
# 3) Save data file to job_upload_dir
data_filename = secure_filename(data_file.filename)
data_path = job_upload_dir / data_filename
try:
data_file.seek(0)
data_file.save(str(data_path))
logger.info(f"Job {job_id}: Data file saved successfully - Path: {data_path}")
except Exception as e:
logger.error(f"Job {job_id}: Failed to save data file {data_filename}: {str(e)}")
return _format_response(500, "保存数据文件失败")
# 4) Save config file to job_upload_dir
if config_file and config_file.filename != '':
config_filename = secure_filename(config_file.filename)
config_path = job_upload_dir / config_filename
try:
config_file.seek(0)
config_file.save(str(config_path))
active_config_path = config_path
logger.info(f"Job {job_id}: Custom config saved successfully - Path: {config_path}")
except Exception as e:
logger.error(f"Job {job_id}: Failed to save config file {config_filename}: {str(e)}")
return _format_response(500, "保存配置文件失败")
else:
# Copy default config for record keeping
config_path = job_upload_dir / "config.yaml"
with open(config_path, 'w', encoding='utf-8') as f:
yaml.safe_dump(active_config, f, allow_unicode=True)
active_config_path = config_path
logger.info(f"Job {job_id}: Default config saved for record - Path: {config_path}")
# Initialize task status
update_task_status(job_id, TASK_STATUS_PENDING, "Task queued for processing")
logger.info(f"Job {job_id}: Task status initialized as PENDING")
# Start background processing
try:
thread = threading.Thread(
target=process_data_async,
args=(job_id, data_path, active_config_path, job_output_dir)
)
thread.daemon = True
thread.start()
logger.info(f"Job {job_id}: Background processing thread started successfully")
except Exception as e:
logger.error(f"Job {job_id}: Failed to start background processing thread: {str(e)}")
update_task_status(job_id, TASK_STATUS_FAILED, error=str(e))
return _format_response(500, "启动处理失败")
logger.info(f"Job {job_id}: Upload process completed successfully, returning job ID to client")
return _format_response(202, "任务已接受并加入处理队列", {
"status": "accepted",
"job_id": job_id,
"task_status_url": f"/task/{job_id}"
})

View File

@ -0,0 +1,233 @@
"""
Web Blueprint
Provides web interface for the GasFlux API.
"""
import time
from pathlib import Path
from flask import Blueprint, render_template_string, current_app
from ..shared import log_performance, logger
from ..app import Config
# Create blueprint
web_bp = Blueprint('web', __name__)
@web_bp.route('/')
@log_performance
def index():
logger.debug("Index page requested")
# 递归查找所有生成的 HTML 报告
start_time = time.time()
all_reports = []
# 优先用 app.config 中的目录,其次回退到 Config.OUTPUT_FOLDER都不存在则不列出文件
output_root = current_app.config.get('OUTPUT_FOLDER') or getattr(Config, 'OUTPUT_FOLDER', None)
if not output_root:
output_path = None
else:
output_path = Path(output_root)
if output_path and output_path.exists():
try:
for file in output_path.rglob("*.html"):
# 获取相对于 OUTPUT_FOLDER 的相对路径,用于下载链接
rel_path = file.relative_to(output_path).as_posix()
all_reports.append(rel_path)
scan_duration = time.time() - start_time
logger.debug(f"Report scan completed in {scan_duration:.3f}s - found {len(all_reports)} HTML reports")
except Exception as e:
logger.error(f"Error scanning for reports: {str(e)}")
all_reports = []
else:
logger.debug("No output directory configured yet, skipping report scan")
all_reports = []
return render_template_string('''
<!doctype html>
<html>
<head>
<title>GasFlux Web API</title>
<style>
body { font-family: sans-serif; margin: 40px; line-height: 1.6; background-color: #f4f7f6; }
.container { max-width: 900px; margin: auto; background: white; padding: 30px; border-radius: 12px; box-shadow: 0 4px 6px rgba(0,0,0,0.1); }
h1 { color: #2c3e50; border-bottom: 2px solid #3498db; padding-bottom: 10px; }
.upload-section { background: #f8f9fa; padding: 25px; border-radius: 8px; border-left: 5px solid #3498db; margin-bottom: 30px; }
.form-group { margin-bottom: 20px; }
label { display: block; font-weight: bold; margin-bottom: 8px; color: #34495e; }
input[type="file"] { display: block; width: 100%; padding: 10px; border: 1px solid #ddd; border-radius: 4px; }
input[type="submit"] { background: #3498db; color: white; border: none; padding: 12px 25px; border-radius: 4px; cursor: pointer; font-size: 16px; transition: background 0.3s; }
input[type="submit"]:hover { background: #2980b9; }
.results-section { margin-top: 40px; }
.report-item { margin-bottom: 15px; padding: 15px; border: 1px solid #eee; border-radius: 6px; display: flex; justify-content: space-between; align-items: center; }
.report-info { display: flex; flex-direction: column; }
.report-link { font-weight: bold; color: #2980b9; text-decoration: none; font-size: 1.1em; }
.report-link:hover { text-decoration: underline; }
.report-path { font-size: 0.85em; color: #7f8c8d; margin-top: 4px; }
.api-docs { margin-top: 50px; padding: 20px; background: #e8f4f8; border-radius: 8px; font-size: 0.9em; }
code { background: #eee; padding: 2px 5px; border-radius: 3px; font-family: monospace; }
</style>
</head>
<body>
<div class="container">
<h1>GasFlux Web API 控制台</h1>
<div class="upload-section">
<h2>新建处理任务</h2>
<form id="uploadForm" enctype=multipart/form-data>
<div class="form-group">
<label for="data_file">数据文件 (Excel):</label>
<input type="file" name="file" id="data_file" required>
</div>
<div class="form-group">
<label for="config_file">配置文件 (YAML) [可选]:</label>
<input type="file" name="config" id="config_file">
</div>
<button type="submit" id="submitBtn">开始上传并分析</button>
</form>
<div id="taskStatus" style="display: none; margin-top: 20px; padding: 15px; background: #e8f8e8; border-radius: 5px; border: 1px solid #28a745;">
<h3>任务状态</h3>
<p id="statusMessage">正在上传文件...</p>
<div id="progressBar" style="width: 100%; height: 20px; background: #f0f0f0; border-radius: 10px; margin: 10px 0; display: none;">
<div id="progressFill" style="height: 100%; background: #28a745; border-radius: 10px; width: 0%; transition: width 0.3s;"></div>
</div>
<p id="taskId" style="font-size: 0.9em; color: #666;"></p>
</div>
</div>
<div class="results-section">
<h2>已生成的报告</h2>
<div id="reports">
{% for report in reports %}
<div class="report-item">
<div class="report-info">
<a class="report-link" href="/download/{{ report }}" target="_blank">{{ report.split('/')[-1] }}</a>
<span class="report-path">任务 ID: {{ report.split('/')[0] }}</span>
</div>
<a href="/download/{{ report }}" download class="report-link" style="font-size: 0.9em;">下载</a>
</div>
{% else %}
<p style="color: #95a5a6;">暂无已生成的报告。</p>
{% endfor %}
</div>
</div>
<div class="api-docs">
<h3>API 调用指南 (开发者)</h3>
<p><strong>健康检查:</strong> <code>GET /health</code></p>
<p><strong>上传分析:</strong> <code>POST /upload</code></p>
<p><strong>查询任务状态:</strong> <code>GET /task/&lt;task_id&gt;</code></p>
<p>参数: <code>file</code> (Excel), <code>config</code> (YAML, 可选)</p>
<p>示例: <code>curl -X POST -F "file=@data.xlsx" http://localhost:5000/upload</code></p>
<p>状态查询: <code>curl http://localhost:5000/task/your-task-id</code></p>
</div>
<script>
document.getElementById('uploadForm').addEventListener('submit', async function(e) {
e.preventDefault();
const formData = new FormData(this);
const submitBtn = document.getElementById('submitBtn');
const taskStatus = document.getElementById('taskStatus');
const statusMessage = document.getElementById('statusMessage');
const taskIdElement = document.getElementById('taskId');
const progressBar = document.getElementById('progressBar');
const progressFill = document.getElementById('progressFill');
// Disable form and show status
submitBtn.disabled = true;
submitBtn.textContent = '上传中...';
taskStatus.style.display = 'block';
progressBar.style.display = 'block';
progressFill.style.width = '10%';
try {
// Upload file
statusMessage.textContent = '正在上传文件...';
const response = await fetch('/upload', {
method: 'POST',
body: formData
});
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`);
}
const result = await response.json();
const taskId = result.data.job_id;
statusMessage.textContent = '文件上传成功,开始处理数据...';
taskIdElement.textContent = `任务ID: ${taskId}`;
progressFill.style.width = '30%';
// Poll for status
let pollCount = 0;
const maxPolls = 300; // 5 minutes max (every 1 second)
const pollStatus = async () => {
try {
const statusResponse = await fetch(`/task/${taskId}`);
const status = await statusResponse.json();
if (status.data.status === 'completed') {
statusMessage.textContent = '处理完成!正在准备下载链接...';
progressFill.style.width = '100%';
submitBtn.textContent = '处理完成!';
submitBtn.disabled = false;
// Reload page to show new reports
setTimeout(() => {
window.location.reload();
}, 2000);
} else if (status.data.status === 'failed') {
statusMessage.textContent = `处理失败: ${status.data.error || '未知错误'}`;
progressFill.style.backgroundColor = '#dc3545';
progressFill.style.width = '100%';
submitBtn.textContent = '处理失败';
submitBtn.disabled = false;
} else {
// Still processing
statusMessage.textContent = status.data.message || '正在处理中...';
const progressPercent = Math.min(30 + (pollCount * 70 / maxPolls), 90);
progressFill.style.width = `${progressPercent}%`;
pollCount++;
if (pollCount < maxPolls) {
setTimeout(pollStatus, 1000);
} else {
statusMessage.textContent = '处理超时,请稍后手动检查状态';
submitBtn.textContent = '处理超时';
submitBtn.disabled = false;
}
}
} catch (error) {
console.error('Status check failed:', error);
statusMessage.textContent = '状态检查失败,请手动刷新页面查看结果';
submitBtn.textContent = '状态检查失败';
submitBtn.disabled = false;
}
};
// Start polling
setTimeout(pollStatus, 1000);
} catch (error) {
console.error('Upload failed:', error);
statusMessage.textContent = `上传失败: ${error.message}`;
progressFill.style.backgroundColor = '#dc3545';
progressFill.style.width = '100%';
submitBtn.textContent = '上传失败';
submitBtn.disabled = false;
}
});
</script>
</div>
</body>
</html>
''', reports=all_reports)

View File

@ -29,13 +29,14 @@ from datetime import datetime
import sys
import os
from collections import Counter
import yaml
try:
from tqdm import tqdm
HAS_TQDM = True
except ImportError:
HAS_TQDM = False
print("⚠️ 未安装tqdm库将不显示进度条。如需进度条请运行: pip install tqdm")
print("WARNING: 未安装tqdm库将不显示进度条。如需进度条请运行: pip install tqdm")
def create_height_bins(heights, bin_size=2.0):
@ -82,7 +83,7 @@ def create_height_bins(heights, bin_size=2.0):
# 导入qiya模块
try:
from .qiya import get_pressure_at_location
print("✅ 成功导入qiya模块")
except ImportError as e:
print(f"❌ 导入qiya模块失败: {e}")
print("请确保GasFlux包结构完整")
@ -104,13 +105,13 @@ def load_excel_data(file_path):
# 读取Excel文件
df = pd.read_excel(file_path)
print(f"✅ 成功读取数据:{len(df)} 行,{len(df.columns)} ")
print(f"SUCCESS: Successfully loaded data: {len(df)} rows, {len(df.columns)} columns")
print(f"列名:{list(df.columns)}")
return df
except Exception as e:
print(f"❌ 读取文件失败: {e}")
print(f"ERROR: Failed to read file: {e}")
sys.exit(1)
@ -132,11 +133,11 @@ def remove_columns(df, columns_to_remove):
missing_columns = [col for col in columns_to_remove if col not in df.columns]
if missing_columns:
print(f"⚠️ 以下列不存在(跳过): {missing_columns}")
print(f"以下列不存在(跳过): {missing_columns}")
if existing_columns:
df = df.drop(columns=existing_columns)
print(f"已删除 {len(existing_columns)}")
print(f"已删除 {len(existing_columns)}")
return df
@ -158,7 +159,7 @@ def extract_hour_from_filename(filename):
if match:
return match.group(1)
else:
print(f"⚠️ 无法从文件名 '{filename}' 中提取小时信息,使用默认值 '00'")
print(f"无法从文件名 '{filename}' 中提取小时信息,使用默认值 '00'")
return "00"
@ -181,7 +182,7 @@ def fix_time_column(df, filename):
# 检查时间列是否存在
if '时间' not in df.columns:
print("未找到 '时间'")
print("未找到 '时间'")
return df
# 修正时间格式
@ -231,18 +232,18 @@ def convert_coordinates(df):
"""
print("转换经纬度坐标...")
if '经度' in df.columns:
original_lon = df['经度'].head(3).tolist()
df['经度'] = df['经度'] / 1e7
converted_lon = df['经度'].head(3).tolist()
if 'stGPSPositionX' in df.columns:
original_lon = df['stGPSPositionX'].head(3).tolist()
df['stGPSPositionX'] = df['stGPSPositionX'] / 1e7
converted_lon = df['stGPSPositionX'].head(3).tolist()
print("经度转换示例:")
for orig, conv in zip(original_lon, converted_lon):
print(".6f")
if '纬度' in df.columns:
original_lat = df['纬度'].head(3).tolist()
df['纬度'] = df['纬度'] / 1e7
converted_lat = df['纬度'].head(3).tolist()
if 'stGPSPositionY' in df.columns:
original_lat = df['stGPSPositionY'].head(3).tolist()
df['stGPSPositionY'] = df['stGPSPositionY'] / 1e7
converted_lat = df['stGPSPositionY'].head(3).tolist()
print("纬度转换示例:")
for orig, conv in zip(original_lat, converted_lat):
print(".6f")
@ -265,42 +266,54 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
"""
print("计算气压数据...")
# 检查必要列是否存在
required_cols = ['日期', '时间', '经度', '纬度', '融合高程']
# 检查必要列是否存在(支持原始列名和新列名)
# 原始数据中的列名
original_cols = ['qStrDate', 'qStrTime', 'stGPSPositionX', 'stGPSPositionY', 'fAltitudeFused']
# 坐标转换后的列名经纬度已被除以1e7
converted_cols = ['qStrDate', 'qStrTime', 'stGPSPositionX', 'stGPSPositionY', 'fAltitudeFused']
# 优先使用转换后的列名(如果存在),否则使用原始列名
date_col = 'qStrDate' if 'qStrDate' in df.columns else '日期'
time_col = 'qStrTime' if 'qStrTime' in df.columns else '时间'
lon_col = 'stGPSPositionX' if 'stGPSPositionX' in df.columns else '经度'
lat_col = 'stGPSPositionY' if 'stGPSPositionY' in df.columns else '纬度'
height_col = 'fAltitudeFused' if 'fAltitudeFused' in df.columns else '融合高程'
required_cols = [date_col, time_col, lon_col, lat_col, height_col]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
print(f"缺少必要列: {missing_cols}")
print(f"缺少必要列: {missing_cols}")
return df
# 检查高度变化范围
height_min = df['融合高程'].min()
height_max = df['融合高程'].max()
height_min = df[height_col].min()
height_max = df[height_col].max()
height_range = height_max - height_min
print(f"🏔️ 高度范围: {height_min:.1f} - {height_max:.1f} 米 (变化: {height_range:.1f} 米)")
print(f"高度范围: {height_min:.1f} - {height_max:.1f} 米 (变化: {height_range:.1f} 米)")
# 创建高度分档
height_bins = create_height_bins(df['融合高程'], height_bin_size)
print(f"📏 高度分档: {len(height_bins)} 个档位 (间隔: {height_bin_size:.1f} 米)")
height_bins = create_height_bins(df[height_col], height_bin_size)
print(f"高度分档: {len(height_bins)} 个档位 (间隔: {height_bin_size:.1f} 米)")
for i, (bin_min, bin_max, bin_center, count) in enumerate(height_bins):
print(f" 档位{i+1}: {bin_min:.1f}-{bin_max:.1f}m (中心: {bin_center:.1f}m, 数据: {count}行)")
# 决定计算策略
if height_range <= height_tolerance:
# 高度变化小,只计算一次气压
print("🎯 高度变化小,将使用平均高度计算一次气压")
print("高度变化小,将使用平均高度计算一次气压")
use_single_calculation = True
mean_height = df['融合高程'].mean()
print(f"📍 使用平均高度: {mean_height:.1f}")
mean_height = df[height_col].mean()
print(f"使用平均高度: {mean_height:.1f}")
elif len(height_bins) == 1:
# 只有一个高度档位,使用档位中心高度
print("📦 只有一个高度档位,使用档位中心高度")
print("只有一个高度档位,使用档位中心高度")
use_single_calculation = True
mean_height = height_bins[0][2] # bin_center
print(f"📍 使用档位中心高度: {mean_height:.1f}")
print(f"使用档位中心高度: {mean_height:.1f}")
else:
# 高度变化大,使用分档计算
print("🏗️ 使用高度分档策略减少API调用")
print("使用高度分档策略减少API调用")
use_single_calculation = False
# 确定要处理的行数
@ -309,10 +322,10 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
sample_df = df.copy()
actual_samples = len(df)
if not use_single_calculation:
print(f"📊 将计算所有 {len(df)} 行的气压数据")
print(f"将计算所有 {len(df)} 行的气压数据")
else:
# 限制采样数量
print(f"⚠️ 数据量较大 ({len(df)} 行),只对前 {max_samples} 行计算气压")
print(f"数据量较大 ({len(df)} 行),只对前 {max_samples} 行计算气压")
sample_df = df.head(max_samples).copy()
actual_samples = max_samples
@ -325,11 +338,11 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
first_row = sample_df.iloc[0]
# 转换日期格式 - 只提取日期部分,移除任何时间信息
date_str = str(first_row['日期'])
date_str = str(first_row[date_col])
if ' ' in date_str:
date_str = date_str.split(' ')[0]
date_str = date_str.split(' ')[0] # 处理 "2026-01-15 00:00:00" 格式
elif 'T' in date_str:
date_str = date_str.split('T')[0]
date_str = date_str.split('T')[0] # 处理ISO格式
if '/' in date_str:
date_str = date_str.replace('/', '-')
@ -343,8 +356,15 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
# 使用数据的代表性时间(整点小时,众数)
time_strings = []
for time_val in sample_df['时间']:
for time_val in sample_df[time_col]:
time_str = str(time_val).strip()
# 处理时间字符串,提取正确的部分
# 如果时间字符串包含多个时间部分(如 "00:00:00 08:37:12"),取最后一个
if ' ' in time_str:
time_parts = time_str.split()
time_str = time_parts[-1] # 取最后一个有效的时间部分
if ':' in time_str:
# 确保是有效的 HH:MM 格式,然后取整点小时
parts = time_str.split(':')
@ -368,8 +388,8 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
print("正在计算平均气压...")
pressure = get_pressure_at_location(
lat=sample_df['纬度'].mean(),
lon=sample_df['经度'].mean(),
lat=sample_df[lat_col].mean(),
lon=sample_df[lon_col].mean(),
altitude=mean_height,
date=formatted_date,
time=formatted_time
@ -377,18 +397,18 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
if pressure is not None:
pressures = [pressure] * len(sample_df)
print("平均气压计算成功,将应用到所有行")
print("平均气压计算成功,将应用到所有行")
else:
print("平均气压计算失败")
print("平均气压计算失败")
pressures = [None] * len(sample_df)
except Exception as e:
print(f"平均气压计算失败: {e}")
print(f"平均气压计算失败: {e}")
pressures = [None] * len(sample_df)
else:
# 使用高度分档策略
print("🏗️ 开始分档计算气压...")
print("开始分档计算气压...")
# 为每个高度档位计算气压
bin_pressures = {} # bin_center -> pressure
@ -402,19 +422,19 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
try:
# 使用第一行数据作为代表来获取日期和时间
# 找到这个档位中的一行数据
bin_rows = sample_df[(sample_df['融合高程'] >= bin_min) &
(sample_df['融合高程'] <= bin_max)]
bin_rows = sample_df[(sample_df[height_col] >= bin_min) &
(sample_df[height_col] <= bin_max)]
if len(bin_rows) == 0:
continue
first_row = bin_rows.iloc[0]
# 转换日期格式 - 只提取日期部分,移除任何时间信息
date_str = str(first_row['日期'])
date_str = str(first_row[date_col])
if ' ' in date_str:
date_str = date_str.split(' ')[0]
date_str = date_str.split(' ')[0] # 处理 "2026-01-15 00:00:00" 格式
elif 'T' in date_str:
date_str = date_str.split('T')[0]
date_str = date_str.split('T')[0] # 处理ISO格式
if '/' in date_str:
date_str = date_str.replace('/', '-')
@ -424,14 +444,21 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
year, month, day = date_parts
formatted_date = f"{year}-{month.zfill(2)}-{day.zfill(2)}"
else:
print(f"⚠️ 档位高度 {bin_center:.1f}m 日期格式异常: {date_str}")
print(f"档位高度 {bin_center:.1f}m 日期格式异常: {date_str}")
bin_pressures[bin_center] = None
continue
# 使用该档位数据的代表性时间(整点小时)
time_strings = []
for time_val in bin_rows['时间']:
for time_val in bin_rows[time_col]:
time_str = str(time_val).strip()
# 处理时间字符串,提取正确的部分
# 如果时间字符串包含多个时间部分(如 "00:00:00 08:37:12"),取最后一个
if ' ' in time_str:
time_parts = time_str.split()
time_str = time_parts[-1] # 取最后一个有效的时间部分
if ':' in time_str:
# 确保是有效的 HH:MM 格式,然后取整点小时
parts = time_str.split(':')
@ -456,8 +483,8 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
print(" 无有效时间数据使用默认中午12:00")
# 计算这个档位的气压(使用平均位置和档位中心高度)
avg_lat = bin_rows['纬度'].mean()
avg_lon = bin_rows['经度'].mean()
avg_lat = bin_rows[lat_col].mean()
avg_lon = bin_rows[lon_col].mean()
pressure = get_pressure_at_location(
lat=avg_lat,
@ -475,14 +502,14 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
iterator.set_description(f"计算档位 (成功: {success_count}/{len(bin_pressures)})")
except Exception as e:
print(f"计算高度档位 {bin_center:.1f}m 气压失败: {e}")
print(f"计算高度档位 {bin_center:.1f}m 气压失败: {e}")
bin_pressures[bin_center] = None
# 为每一行分配对应档位的气压
pressures = []
for idx, row in sample_df.iterrows():
# 找到这个高度对应的档位
height = row['融合高程']
height = row[height_col]
assigned_pressure = None
for bin_min, bin_max, bin_center, count in height_bins:
@ -492,7 +519,7 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
pressures.append(assigned_pressure)
print(f"完成分档气压计算,共 {len(bin_pressures)} 个档位,{len([p for p in bin_pressures.values() if p is not None])} 个成功")
print(f"完成分档气压计算,共 {len(bin_pressures)} 个档位,{len([p for p in bin_pressures.values() if p is not None])} 个成功")
# 添加气压列
df['pressure'] = None # 初始化
@ -513,7 +540,7 @@ def calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_s
avg_pressure = sum(valid_pressures) / len(valid_pressures)
print(f"成功计算 {len(valid_pressures)}/{actual_samples} 个气压值,平均值: {avg_pressure:.1f} hPa")
else:
print("⚠️ 未能计算出任何气压值")
print("未能计算出任何气压值")
return df
@ -529,13 +556,13 @@ def adjust_altitude(df):
"""
print("调整融合高程...")
if '融合高程' in df.columns:
min_altitude = df['融合高程'].min()
if 'fAltitudeFused' in df.columns:
min_altitude = df['fAltitudeFused'].min()
print(".2f")
original_alt = df['融合高程'].head(3).tolist()
df['融合高程'] = df['融合高程'] - min_altitude
adjusted_alt = df['融合高程'].head(3).tolist()
original_alt = df['fAltitudeFused'].head(3).tolist()
df['fAltitudeFused'] = df['fAltitudeFused'] - min_altitude
adjusted_alt = df['fAltitudeFused'].head(3).tolist()
print("高度调整示例:")
for orig, adj in zip(original_alt, adjusted_alt):
@ -546,7 +573,7 @@ def adjust_altitude(df):
def merge_timestamp(df):
"""
融合日期和时间列为时间戳
融合日期和时间列为时间戳(修正时间格式)
Args:
df: 输入DataFrame
@ -554,57 +581,48 @@ def merge_timestamp(df):
Returns:
pd.DataFrame: 融合后的DataFrame
"""
print("融合日期和时间...")
print("融合日期和时间(修正时间格式)...")
if '日期' in df.columns and '时间' in df.columns:
if 'qStrDate' in df.columns and 'qStrTime' in df.columns:
timestamps = []
for idx, row in df.iterrows():
try:
date_str = str(row['日期'])
time_str = str(row['时间'])
date_str = str(row['qStrDate']).strip()
time_str = str(row['qStrTime']).strip()
# 理日期字符串 - 移除任何时间部分
date_str = date_str.strip()
# 理日期字符串,提取纯日期部分
# 如果日期字符串包含时间部分(如 "2026-01-15 00:00:00"),取日期部分
if ' ' in date_str:
date_str = date_str.split(' ')[0] # 只取日期部分
if 'T' in date_str:
date_str = date_str.split('T')[0] # 处理ISO格式
date_parts = date_str.split()
date_str = date_parts[0] # 取第一个部分作为日期
# 标准化日期格式
if '/' in date_str:
date_str = date_str.replace('/', '-')
# 处理时间字符串,提取正确的部分
# 如果时间字符串包含多个时间部分(如 "00:00:00 08:37:12"),取最后一个
if ' ' in time_str:
time_parts = time_str.split()
# 取最后一个有效的时间部分
time_str = time_parts[-1]
# 确保日期格式正确
date_parts = date_str.split('-')
if len(date_parts) == 3:
year, month, day = date_parts
date_formatted = f"{year.zfill(4)}-{month.zfill(2)}-{day.zfill(2)}"
else:
print(f"⚠️ 日期格式异常: '{date_str}',使用当前日期")
date_formatted = datetime.now().strftime("%Y-%m-%d")
# 时间字符串已经是修正后的格式(如 "08:34:01"),直接使用
time_str = time_str.strip()
# 确保时间格式正确
if ':' in time_str and len(time_str.split(':')) >= 2:
time_formatted = time_str
# 组合日期和修正后的时间
timestamp = f"{date_str} {time_str}"
else:
print(f"⚠️ 时间格式异常: '{time_str}',使用默认时间")
time_formatted = "12:00:00"
timestamp = f"{date_str} 12:00:00" # 默认中午时间
print(f" 时间格式异常 '{row['qStrTime']}',使用默认时间")
# 组合时间戳 - 直接连接日期和时间
timestamp = f"{date_formatted} {time_formatted}"
timestamps.append(timestamp)
except Exception as e:
print(f"处理第 {idx+1} 行时间戳失败: {e}")
print(f"处理第 {idx+1} 行时间戳失败: {e}")
timestamps.append(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
df['timestamp'] = timestamps
print("时间戳融合示例:")
for i in range(min(3, len(timestamps))):
print(f" {df.loc[i, '日期']} + {df.loc[i, '时间']}{timestamps[i]}")
print(f" {df.loc[i, 'qStrDate']} + {df.loc[i, 'qStrTime']}{timestamps[i]}")
return df
@ -623,83 +641,104 @@ def rename_columns(df):
# 定义字段映射
column_mapping = {
'timestamp': 'timestamp', # 时间戳(已创建)
'经度': 'longitude', # 经度 → latitude
'纬度': 'latitude', # 纬度 → longitude
'融合高程': 'height_ato', # 融合高程 → height_ato
'修正风向': 'winddir', # 修正风向 → winddir
'修正风速': 'windspeed', # 修正风速 → windspeed
'风温': 'temperature', # 风温 → temperature
'pressure': 'pressure', # 气压(已计算)
'CH4': 'ch4', # CH4保持不变
'pitch': 'course_elevation', # pitch → course_elevation
'yaw': 'course_azimuth' # yaw → course_azimuth
'timestamp': 'timestamp', # 时间戳(已创建)
'stGPSPositionX': 'longitude', # 经度 → longitude
'stGPSPositionY': 'latitude', # 纬度 → latitude
'fAltitudeFused': 'height_ato', # 融合高程 → height_ato
'fFixedWindDirection': 'winddir', # 修正风向 → winddir
'fFixedWindSpeed': 'windspeed', # 修正风速 → windspeed
'fWindTemperature': 'temperature', # 风温 → temperature
'CO2': 'CO2', # CO2浓度 → co2
'pitch': 'course_elevation', # pitch → course_elevation
'yaw': 'course_azimuth' # yaw → course_azimuth
}
# 重命名存在的列
columns_to_rename = {}
# 先复制字段,再对复制的字段重命名
columns_renamed = []
for old_name, new_name in column_mapping.items():
if old_name in df.columns:
columns_to_rename[old_name] = new_name
# 复制字段到新名称
df[new_name] = df[old_name].copy()
columns_renamed.append((old_name, new_name))
print(f" 复制并重命名: {old_name}{new_name}")
if columns_to_rename:
df = df.rename(columns=columns_to_rename)
print("字段重命名:")
for old, new in columns_to_rename.items():
print(f" {old}{new}")
# 只保留GasFlux需要的列
required_columns = ['timestamp', 'latitude', 'longitude', 'height_ato', 'windspeed', 'winddir', 'temperature', 'pressure', 'ch4', 'course_elevation', 'course_azimuth']
existing_required_columns = [col for col in required_columns if col in df.columns]
if len(existing_required_columns) != len(required_columns):
missing = [col for col in required_columns if col not in df.columns]
print(f"⚠️ 缺少必需列: {missing}")
# 移除不需要的列,只保留必需的列
df = df[existing_required_columns]
print(f"最终保留列: {existing_required_columns}")
if columns_renamed:
print(f"共处理了 {len(columns_renamed)} 个字段")
return df
def process_excel_file(file_path):
def ensure_float64_types(df, config=None):
"""
确保数值字段为float64类型以满足GasFlux处理要求
Args:
df: 输入DataFrame
config: 配置字典包含gases字段
Returns:
pd.DataFrame: 数据类型转换后的DataFrame
"""
print("确保数值字段类型为float64...")
# 定义基础需要转换为float64的字段
float64_columns = [
'longitude', 'latitude', 'height_ato', # 位置和高度
'winddir', 'windspeed', 'temperature', # 风和温度
'pressure', # 气压
'course_elevation', 'course_azimuth' # 姿态角
]
# 如果提供了配置添加gases中的气体列
if config and 'gases' in config:
gas_columns = list(config['gases'].keys())
float64_columns.extend(gas_columns)
print(f"从配置中添加气体列: {gas_columns}")
converted_count = 0
for col in float64_columns:
if col in df.columns:
try:
original_dtype = df[col].dtype
df[col] = df[col].astype('float64')
new_dtype = df[col].dtype
if original_dtype != new_dtype:
print(f" 转换: {col} ({original_dtype}{new_dtype})")
converted_count += 1
except Exception as e:
print(f" 转换失败: {col} - {e}")
if converted_count > 0:
print(f"共转换了 {converted_count} 个字段的数据类型为float64")
else:
print("所有数值字段已经是float64类型")
return df
def process_excel_file(file_path, config_path=None):
"""
处理单个Excel文件的主函数
Args:
file_path: Excel文件路径
config_path: 配置文件路径用于读取gases配置
"""
print(f"=== 开始处理文件: {file_path} ===\n")
# 获取文件名(用于时间修正)
filename = Path(file_path).name
# 1. 读取数据
df = load_excel_data(file_path)
# 2. 删除不需要的列
columns_to_remove = [
'高程', '速度x', '速度y', '速度z',
'四元数_q0', '四元数_q1', '四元数_q2', '四元数_q3',
'roll', 'H2O', # 保留pitch和yaw将重命名为course_elevation和course_azimuth
'原始风向', '原始风速'
]
df = remove_columns(df, columns_to_remove)
# 3. 修正时间格式
df = fix_time_column(df, filename)
# 4. 坐标转换
# 2. 坐标转换
df = convert_coordinates(df)
# 5. 计算气压
# 4. 计算气压数据
df = calculate_pressure(df, max_samples=None, height_tolerance=10.0, height_bin_size=2.0) # 计算所有行高度容差10米分档2米
# 6. 高度调整
# 5. 高度调整
df = adjust_altitude(df)
# 7. 时间戳融合
# 6. 时间戳融合(保持原有时间格式)
df = merge_timestamp(df)
# 调试:检查当前列
@ -707,28 +746,41 @@ def process_excel_file(file_path):
if 'timestamp' in df.columns:
print(f"timestamp列示例: {df['timestamp'].head(3).tolist()}")
# 8. 字段重命名
# 7. 字段重命名
df = rename_columns(df)
# 8. 确保数值字段类型为float64
# 读取配置以获取gases字段
config = None
if config_path:
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
except Exception as e:
print(f"读取配置文件失败: {e}")
df = ensure_float64_types(df, config)
# 保存处理结果
output_path = Path(file_path).with_suffix('.processed.csv')
df.to_csv(output_path, index=False)
df.to_csv(output_path, index=False, encoding='utf-8-sig')
print(f"\n处理完成!")
print(f"📁 输出文件: {output_path}")
print(f"📊 最终数据形状: {df.shape[0]}× {df.shape[1]}")
print(f"📋 最终列名: {list(df.columns)}")
print(f"\n处理完成!")
print(f"输出文件: {output_path}")
print(f"最终数据形状: {df.shape[0]}× {df.shape[1]}")
print(f"最终列名: {list(df.columns)}")
return df
def process_file(input_file, output_file=None):
def process_file(input_file, output_file=None, config_file=None):
"""
直接处理Excel文件的函数不使用命令行参数
Args:
input_file: 输入Excel文件路径字符串或Path对象
output_file: 输出CSV文件路径可选字符串或Path对象
config_file: 配置文件路径可选用于读取gases配置
Returns:
pd.DataFrame: 处理后的DataFrame
@ -744,13 +796,13 @@ def process_file(input_file, output_file=None):
raise ValueError(f"输入文件必须是Excel格式 (.xlsx 或 .xls),当前文件: {input_path}")
# 处理文件
df = process_excel_file(str(input_path))
df = process_excel_file(str(input_path), config_file)
# 如果指定了输出路径,额外保存一份
if output_file:
output_path = Path(output_file)
df.to_csv(output_path, index=False)
print(f"📁 额外保存到: {output_path}")
df.to_csv(output_path, index=False, encoding='utf-8-sig')
print(f"额外保存到: {output_path}")
return df
@ -769,17 +821,17 @@ def interactive_input():
while True:
input_file = input("请输入Excel文件路径 (例如: data.xlsx): ").strip()
if not input_file:
print("文件路径不能为空,请重新输入")
print("文件路径不能为空,请重新输入")
continue
input_path = Path(input_file)
if not input_path.exists():
print(f"文件不存在: {input_path}")
print(f"文件不存在: {input_path}")
print("提示: 请确保文件路径正确,或者将文件放在当前目录下")
continue
if input_path.suffix.lower() not in ['.xlsx', '.xls']:
print(f"文件格式错误: {input_path.suffix}")
print(f"文件格式错误: {input_path.suffix}")
print("只支持 .xlsx 和 .xls 格式的Excel文件")
continue
@ -791,13 +843,13 @@ def interactive_input():
output_file = None
print("使用默认输出文件名")
print(f"\n输入确认:")
print(f"\n输入确认:")
print(f" 输入文件: {input_file}")
print(f" 输出文件: {output_file or '自动生成'}")
confirm = input("\n确认开始处理? (y/N): ").strip().lower()
if confirm not in ['y', 'yes', '', '确认']:
print("用户取消操作")
print("用户取消操作")
return None, None
return input_file, output_file
@ -823,7 +875,7 @@ def main(input_file=None, output_file=None, interactive=False):
try:
return process_file(input_file, output_file)
except Exception as e:
print(f"处理失败: {e}")
print(f"处理失败: {e}")
raise
# 否则使用命令行参数
@ -881,7 +933,7 @@ def main(input_file=None, output_file=None, interactive=False):
if not input_file:
parser.print_help()
print("\n" + "="*60)
print("📖 使用示例:")
print("使用示例:")
print("="*60)
print("1. 命令行模式:")
print(" python data_processor.py your_file.xlsx")
@ -900,11 +952,11 @@ def main(input_file=None, output_file=None, interactive=False):
# 检查输入文件
input_path = Path(input_file)
if not input_path.exists():
print(f"错误:输入文件不存在: {input_path}")
print(f"错误:输入文件不存在: {input_path}")
sys.exit(1)
if input_path.suffix.lower() not in ['.xlsx', '.xls']:
print(f"错误输入文件必须是Excel格式 (.xlsx 或 .xls)")
print(f"错误输入文件必须是Excel格式 (.xlsx 或 .xls)")
sys.exit(1)
# 处理文件
@ -914,16 +966,16 @@ def main(input_file=None, output_file=None, interactive=False):
# 如果指定了输出路径,额外保存一份
if output_file:
output_path = Path(output_file)
df.to_csv(output_path, index=False)
print(f"📁 额外保存到: {output_path}")
df.to_csv(output_path, index=False, encoding='utf-8-sig')
print(f"额外保存到: {output_path}")
return df
except KeyboardInterrupt:
print("\n⚠️ 用户中断处理")
print("\n用户中断处理")
sys.exit(1)
except Exception as e:
print(f"\n处理失败: {e}")
print(f"\n处理失败: {e}")
import traceback
traceback.print_exc()
sys.exit(1)

View File

@ -192,8 +192,13 @@ class SpiralSpatialProcessingStrategy(SpatialProcessingStrategy):
self.data_processor.circle_center_x,
self.data_processor.circle_center_y,
) = processing.circle_deviation(self.data_processor.df, x_col="utm_easting", y_col="utm_northing")
# 使用配置中第一个气体的归一化列名
primary_gas = self.data_processor.gases[0]
y_col = f"{primary_gas}_normalised"
self.data_processor.df = processing.recentre_azimuth(
self.data_processor.df, r=self.data_processor.circle_radius
self.data_processor.df, r=self.data_processor.circle_radius, y=y_col
)
self.data_processor.df["x"] = self.data_processor.df["circumference_distance"]
for gas_name in self.data_processor.gases:
@ -304,10 +309,11 @@ class DataProcessor:
].std()
def process_main(data_file: Path, config_file: Path) -> None:
def process_main(data_file: Path, config_file: Path, task_id: str | None = None) -> None:
"""Main function to run the pipeline."""
config = load_config(config_file)
name = data_file.stem
# 优先使用 task_id否则退回文件 stem
name = task_id if task_id else data_file.stem
df = read_csv(data_file)
processor = DataProcessor(config, df)

View File

@ -66,47 +66,52 @@ def generate_reports(name: str, processor, config: dict):
Generates reports, configuration files, and processed output variables for gasflux processing runs.
Parameters:
name (str): The name identifier for the current processing run.
name (str): The name identifier for the current processing run (task_id).
processor (object): The processing object containing report data and output variables.
config (dict): Configuration dictionary used for processing.
"""
output_dir = Path(config["output_dir"]).expanduser()
processing_time = datetime.now()
output_path = output_dir / name / processing_time.strftime("%Y-%m-%d_%H-%M-%S-%f_processing_run")
# Save directly to outputs/{task_id} directory
output_path = output_dir / "outputs" / name
output_path.mkdir(parents=True, exist_ok=True)
# Save reports
for gas, report in processor.reports.items():
report_path = output_path / f"{name}_{gas}_report.html"
timestamp_str = processing_time.strftime("%Y%m%d_%H%M%S")
report_path = output_path / f"{gas}_report_{timestamp_str}.html"
with open(report_path, "w", encoding="utf-8") as file:
file.write(report)
# Save config
header = f"# Gasflux output config for file {name} from processing run at {processing_time}\n"
config_path = output_path / f"{name}_config.yaml"
header = f"# Gasflux output config for task {name} from processing run at {processing_time}\n"
timestamp_str = processing_time.strftime("%Y%m%d_%H%M%S")
config_path = output_path / f"config_{timestamp_str}.yaml"
with open(config_path, "w") as file:
file.write(header)
yaml.safe_dump(config, file)
# Save DataFrame to CSV
# Save DataFrame to Excel
if hasattr(processor, 'df') and processor.df is not None:
csv_path = output_path / f"{name}_data.csv"
processor.df.to_csv(csv_path, index=False)
logger.info(f"DataFrame saved to {csv_path}")
timestamp_str = processing_time.strftime("%Y%m%d_%H%M%S")
excel_path = output_path / f"processed_data_{timestamp_str}.xlsx"
processor.df.to_excel(excel_path, index=False, engine='openpyxl')
logger.info(f"DataFrame saved to {excel_path}")
# Save output variables
output_vars = processor.output_vars
# output_vars = delete_large_arrays(output_vars, threshold_size=50)
header = (
f"# Gasflux output variables for file {name} from processing run at {processing_time}\n"
f"# Gasflux output variables for task {name} from processing run at {processing_time}\n"
)
filename = output_path / f"{name}_output_vars.json"
timestamp_str = processing_time.strftime("%Y%m%d_%H%M%S")
filename = output_path / f"output_vars_{timestamp_str}.json"
with open(filename, "w") as file:
file.write(header)
json.dump(
output_vars, file, default=lambda item: item.tolist() if isinstance(item, np.ndarray) else item, indent=4
)
logger.info(f"Processing run saved to {output_path}")
logger.info(f"Task {name} results saved to {output_path}")
def delete_large_arrays(output_vars: dict, threshold_size: int) -> dict:

View File

@ -81,7 +81,7 @@ def main():
if args.output:
processed_csv = Path(args.output)
else:
processed_csv = input_path.with_suffix('.processed.csv')
processed_csv = input_path.with_suffix('_processed.csv')
# 确定配置文件
if args.config:
@ -98,7 +98,7 @@ def main():
try:
# 第一步:数据预处理
print("🔄 第一步:数据预处理...")
processed_df = process_file(str(input_path), str(processed_csv))
processed_df = process_file(str(input_path), str(processed_csv), str(config_file))
print(f"✅ 数据预处理完成,输出文件: {processed_csv}")
print()

704
src/gasflux/shared.py Normal file
View File

@ -0,0 +1,704 @@
"""
Shared utilities and constants for GasFlux API.
This module contains shared functions and variables to avoid circular imports.
"""
import time
import logging
from functools import wraps
from flask import request, current_app, g
from pathlib import Path
import threading
# Task status constants
TASK_STATUS_PENDING = "pending"
TASK_STATUS_PROCESSING = "processing"
TASK_STATUS_COMPLETED = "completed"
TASK_STATUS_FAILED = "failed"
# Global task status storage
task_status = {}
# Task status persistence file override (set by app on startup)
_TASK_STATUS_FILE_PATH: Path | None = None
_TASK_STATUS_FILE_LOCK = threading.Lock()
def set_task_status_file_path(path: str | Path):
"""Set the task status persistence file path (used across threads without Flask app context)."""
global _TASK_STATUS_FILE_PATH
_TASK_STATUS_FILE_PATH = Path(path)
def _build_simple_downloads_from_results(results: list[dict]) -> dict:
"""Build direct download shortcuts for common files (minimal, frontend-friendly)."""
downloads: dict = {}
def set_once(key: str, url: str):
if key not in downloads and url:
downloads[key] = url
for item in results or []:
if not isinstance(item, dict):
continue
rel_path = item.get('rel_path')
if not rel_path:
continue
name_l = (item.get('name') or '').lower()
url = f"/download/{rel_path}"
if name_l.endswith('.xlsx'):
set_once('data_xlsx', url)
elif name_l.endswith('.xls'):
set_once('data_xls', url)
elif name_l.endswith('ch4_report.html'):
set_once('report_ch4', url)
elif name_l.endswith('co2_report.html'):
set_once('report_co2', url)
elif name_l.endswith(('.yaml', '.yml')):
set_once('config', url)
elif name_l.endswith('.json') and 'output_vars' in name_l:
set_once('metadata', url)
elif name_l.endswith('.html'):
set_once('report_html', url)
return downloads
# File extension constants
ALLOWED_DATA_EXTENSIONS = {'xlsx', 'xls'}
ALLOWED_CONFIG_EXTENSIONS = {'yaml', 'yml'}
# Statistics collector
class StatisticsCollector:
"""Collects and manages API statistics."""
def __init__(self):
self.start_time = time.time()
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
self.stats['tasks']['by_status']['completed'] += 1
elif new_status == TASK_STATUS_FAILED:
self.stats['tasks']['total_failed'] += 1
self.stats['tasks']['by_status']['failed'] += 1
elif new_status == TASK_STATUS_PROCESSING:
self.stats['tasks']['by_status']['processing'] += 1
elif new_status == TASK_STATUS_PENDING:
self.stats['tasks']['by_status']['pending'] += 1
def record_task_completion_time(self, completion_time):
"""Record task completion time."""
self.stats['tasks']['processing_times'].append(completion_time)
def reset_stats(self):
"""Reset all statistics."""
current_time = time.time()
self.start_time = current_time
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': current_time - self.start_time
}
}
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)
# Create global statistics collector instance
stats_collector = StatisticsCollector()
# Shared utility functions
def log_performance(func):
"""Decorator to log function performance."""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
duration = time.time() - start_time
logger.debug(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
def log_request_info():
"""Log incoming request information."""
if hasattr(request, 'remote_addr'):
logger.info(f"REQUEST: {request.method} {request.url} - IP: {request.remote_addr} - User-Agent: {request.headers.get('User-Agent', 'Unknown')}")
def log_response_info(response):
"""Log outgoing response information."""
if hasattr(request, 'url_rule') and request.url_rule:
endpoint = request.url_rule.rule
else:
endpoint = request.path
start_time = getattr(g, 'start_time', None)
if start_time:
duration = time.time() - start_time
logger.info(f"RESPONSE: {request.method} {endpoint} - Status: {response.status_code} - Duration: {duration:.3f}s" if duration else f"RESPONSE: {request.method} {endpoint} - Status: {response.status_code}")
def update_task_status(task_id, status, message=None, results=None, error=None):
"""Update task status in the global dictionary."""
timestamp = time.time()
old_status = task_status.get(task_id, {}).get("status", "unknown")
task_status[task_id] = {
"status": status,
"message": message,
"results": results,
"error": error,
"updated_at": timestamp,
"created_at": task_status.get(task_id, {}).get("created_at", timestamp)
}
# Record status change in statistics
stats_collector.record_task_status_change(old_status, status)
# Log detailed status change with context
log_msg = f"Task {task_id} status changed: {old_status} -> {status}"
if message:
log_msg += f" | Message: {message}"
if results:
log_msg += f" | Results count: {len(results) if isinstance(results, list) else 'N/A'}"
if error:
log_msg += f" | Error: {error}"
log_level = logging.ERROR if status == TASK_STATUS_FAILED else logging.INFO
logger.log(log_level, log_msg)
# Save task status to file for persistence
logger.debug(f"Saving task {task_id} status '{status}' to persistent storage")
save_task_status_to_file()
def get_task_status(task_id):
"""Get task status from global dictionary."""
if task_id in task_status:
return task_status[task_id]
return {"status": "not_found"}
def cleanup_old_tasks():
"""Clean up old completed tasks to prevent memory leak."""
current_time = time.time()
max_age = 24 * 3600 # 24 hours
to_remove = []
for task_id, task_info in task_status.items():
task_age = current_time - task_info.get("updated_at", 0)
if task_age > max_age:
to_remove.append(task_id)
logger.info(f"Task {task_id} scheduled for cleanup (age: {task_age:.1f}s, status: {task_info.get('status')})")
initial_count = len(task_status)
for task_id in to_remove:
del task_status[task_id]
if to_remove:
logger.info(f"Cleanup completed: removed {len(to_remove)} tasks, {len(task_status)} tasks remaining")
else:
logger.debug(f"Cleanup check: no old tasks to remove ({len(task_status)} active tasks)")
def get_task_list(status_filter=None, page=1, page_size=20, sort_by='updated_at', sort_order='desc', cleanup=True):
"""
Get paginated list of tasks with optional filtering and sorting.
Args:
status_filter (str or list): Filter by task status. Can be single status or list of statuses.
page (int): Page number (1-based).
page_size (int): Number of tasks per page.
sort_by (str): Sort field ('created_at', 'updated_at', 'status').
sort_order (str): Sort order ('asc' or 'desc').
cleanup (bool): Whether to cleanup old tasks before returning list.
Returns:
dict: {
'tasks': list of task summaries,
'total': total number of tasks,
'page': current page,
'page_size': page size,
'total_pages': total pages,
'has_next': has next page,
'has_prev': has previous page
}
"""
if cleanup:
cleanup_old_tasks() # Clean up old tasks before returning list
# Filter tasks
filtered_tasks = []
for task_id, task_info in task_status.items():
if status_filter:
if isinstance(status_filter, str):
if task_info.get('status') != status_filter:
continue
elif isinstance(status_filter, list):
if task_info.get('status') not in status_filter:
continue
filtered_tasks.append((task_id, task_info))
# Sort tasks
def safe_numeric_sort(value):
"""Safely convert value to numeric for sorting."""
try:
if isinstance(value, (int, float)):
return value
elif isinstance(value, str):
# Try to convert string timestamp to float
return float(value)
else:
return 0
except (ValueError, TypeError):
return 0
reverse_order = sort_order.lower() == 'desc'
if sort_by == 'created_at':
filtered_tasks.sort(key=lambda x: safe_numeric_sort(x[1].get('created_at', 0)), reverse=reverse_order)
elif sort_by == 'updated_at':
filtered_tasks.sort(key=lambda x: safe_numeric_sort(x[1].get('updated_at', 0)), reverse=reverse_order)
elif sort_by == 'status':
filtered_tasks.sort(key=lambda x: x[1].get('status', ''), reverse=reverse_order)
else:
# Default sort by updated_at desc
filtered_tasks.sort(key=lambda x: safe_numeric_sort(x[1].get('updated_at', 0)), reverse=True)
# Paginate
total_tasks = len(filtered_tasks)
total_pages = (total_tasks + page_size - 1) // page_size
start_idx = (page - 1) * page_size
end_idx = start_idx + page_size
paginated_tasks = filtered_tasks[start_idx:end_idx]
# Format task summaries
task_summaries = []
for task_id, task_info in paginated_tasks:
summary = {
'task_id': task_id,
'status': task_info.get('status'),
'message': task_info.get('message'),
'created_at': task_info.get('created_at'),
'updated_at': task_info.get('updated_at'),
'has_results': bool(task_info.get('results')),
'has_error': bool(task_info.get('error'))
}
task_summaries.append(summary)
return {
'tasks': task_summaries,
'total': total_tasks,
'page': page,
'page_size': page_size,
'total_pages': total_pages,
'has_next': page < total_pages,
'has_prev': page > 1
}
def get_task_pool_stats():
"""
Get task pool statistics.
Returns:
dict: Task pool statistics including counts by status.
"""
# Note: cleanup_old_tasks() is disabled for task pool stats
# to preserve historical task data for management purposes
# cleanup_old_tasks() # Clean up old tasks before calculating stats
stats = {
'total_tasks': len(task_status),
'status_counts': {},
'active_tasks': 0,
'queued_tasks': 0,
'completed_tasks': 0,
'failed_tasks': 0
}
for task_info in task_status.values():
status = task_info.get('status')
if status:
stats['status_counts'][status] = stats['status_counts'].get(status, 0) + 1
if status == TASK_STATUS_PROCESSING:
stats['active_tasks'] += 1
elif status == TASK_STATUS_PENDING:
stats['queued_tasks'] += 1
elif status == TASK_STATUS_COMPLETED:
stats['completed_tasks'] += 1
elif status == TASK_STATUS_FAILED:
stats['failed_tasks'] += 1
return stats
def _get_status_file():
"""Get the task status file path, preferring OUTPUT_FOLDER for reliability."""
if _TASK_STATUS_FILE_PATH is not None:
return _TASK_STATUS_FILE_PATH
try:
# Prefer OUTPUT_FOLDER which is more reliable and writable
from flask import current_app
output_dir = current_app.config.get('OUTPUT_FOLDER')
if output_dir:
return Path(output_dir) / "task_status.json"
except Exception:
pass
# Fall back to module directory (for development)
return Path(__file__).parent / "task_status.json"
def _to_json_safe(obj):
"""Convert numpy types and other non-JSON-serializable objects to JSON-safe types."""
# Preserve JSON-native types
if obj is None or isinstance(obj, (str, int, float, bool)):
return obj
# Recursively handle containers
if isinstance(obj, list):
return [_to_json_safe(v) for v in obj]
if isinstance(obj, tuple):
return [_to_json_safe(v) for v in obj]
if isinstance(obj, dict):
return {str(k): _to_json_safe(v) for k, v in obj.items()}
try:
import numpy as np
if isinstance(obj, (np.integer,)):
return int(obj)
if isinstance(obj, (np.floating,)):
return float(obj)
if isinstance(obj, (np.bool_,)):
return bool(obj)
if isinstance(obj, (np.ndarray,)):
return obj.tolist()
except Exception:
pass
# Non-builtin/unknown types, fall back to string representation
try:
return str(obj)
except Exception:
return None
def save_task_status_to_file():
"""Save current task status to JSON file for persistence."""
try:
import json
import os
from pathlib import Path
status_file = _get_status_file()
logger.debug(f"Attempting to save {len(task_status)} task statuses to {status_file}")
# Guard: 避免用空内存覆盖已有文件
if not task_status and status_file.exists():
logger.info("Skipping task status save: empty in-memory status and file already exists")
return
# Ensure only one thread writes the status file at a time (prevents Windows replace/lock issues)
with _TASK_STATUS_FILE_LOCK:
status_to_save = {}
for task_id, task_info in task_status.items():
try:
clean_info = {}
for k, v in task_info.items():
if k == 'results' and isinstance(v, list):
# Keep only metadata for results, remove potentially large data fields
cleaned_results = []
for item in v:
if isinstance(item, dict):
cleaned_item = {kk: vv for kk, vv in item.items() if kk not in ['data', 'arrays']}
# Normalize result metadata for persistence
name = cleaned_item.get('name') or ''
rel_path = cleaned_item.get('rel_path') or ''
# Normalize size to int
size = cleaned_item.get('size', 0)
try:
size = int(size)
except Exception:
size = 0
cleaned_item['size'] = size
# Infer/normalize type if missing or unknown
t = cleaned_item.get('type')
if (not t) or (t == 'unknown'):
t = _get_file_type(name or rel_path)
cleaned_item['type'] = t
# Convert numpy types / other objects to JSON-safe
for kk, vv in list(cleaned_item.items()):
cleaned_item[kk] = _to_json_safe(vv)
cleaned_results.append(cleaned_item)
clean_info['results'] = cleaned_results
# Slim persistence: store only direct downloads for frontend
clean_info['downloads'] = _build_simple_downloads_from_results(cleaned_results)
# Ensure each result has a download_url (optional but handy)
for r in clean_info['results']:
if isinstance(r, dict) and r.get('rel_path') and not r.get('download_url'):
r['download_url'] = f"/download/{r['rel_path']}"
else:
clean_info[k] = _to_json_safe(v)
status_to_save[task_id] = clean_info
logger.debug(f"Processed task {task_id} with status {clean_info.get('status')}")
except Exception as task_e:
logger.warning(f"Failed to process task {task_id}: {task_e}")
# Skip this task but continue with others
continue
# Ensure the directory exists
status_file.parent.mkdir(parents=True, exist_ok=True)
# Write to a unique temporary file first, then rename for atomicity with fsync
temp_file = status_file.with_suffix(f'.{os.getpid()}.tmp')
with open(temp_file, 'w', encoding='utf-8') as f:
json.dump(status_to_save, f, indent=2, ensure_ascii=False)
f.flush()
os.fsync(f.fileno()) # Force write to disk
# Atomic rename
temp_file.replace(status_file)
logger.info(f"Successfully saved {len(status_to_save)} task statuses to {status_file}")
except Exception as e:
logger.error(f"Failed to save task status to file: {e}", exc_info=True)
def load_task_status_from_file():
"""Load task status from JSON file on startup."""
try:
import json
status_file = _get_status_file()
if not status_file.exists():
logger.info("No task status file found, starting with empty task pool")
return
# Ensure no writer thread is updating the file while we read it
with _TASK_STATUS_FILE_LOCK:
with open(status_file, 'r', encoding='utf-8') as f:
loaded_status = json.load(f)
# Restore task status
global task_status
task_status.clear()
task_status.update(loaded_status)
logger.info(f"Loaded {len(loaded_status)} task statuses from {status_file}")
except Exception as e:
logger.warning(f"Failed to load task status from file: {e}")
def allowed_file(filename, allowed_extensions):
"""Check if file extension is allowed."""
return '.' in filename and \
filename.rsplit('.', 1)[1].lower() in allowed_extensions
def _format_response(code, message, data=None):
"""Format API response in unified format."""
response = {
"code": code,
"message": message,
"data": data if data is not None else {}
}
return response, code
def _get_file_type(filename):
"""Determine file type from filename."""
fn = (filename or '').lower()
if fn.endswith('.html'):
return 'report'
elif fn.endswith(('.csv', '.xlsx', '.xls')):
return 'data'
elif fn.endswith('.yaml') or fn.endswith('.yml'):
return 'config'
elif fn.endswith('.json'):
return 'metadata'
else:
return 'other'
# Initialize logger
logger = logging.getLogger('gasflux_api')