Metrix/InterferenceETL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 修正CGI并固定数据库配置

Browse Source
This commit is contained in:
nixevol
2026-07-31 17:27:20 +08:00
parent ad50be3256
commit e063ed0692
5 changed files with 64 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+26 -67
View File
@@ -38,6 +38,14 @@ DEFAULT_STORAGE_ID = "stg_4d9a910d72"
DEFAULT_ROOT = "/网优日常优化数据文档/(勿删)干扰定时小时指标"
FILE_RE = re.compile(r"^(?P<source_type>.+)_LWP_每小时_过滤110_(?P<window>\d{16})\.zip$", re.IGNORECASE)
LTE_PLMN = "460-00"
DATABASE_HOST = "172.17.0.1"
DATABASE_PORT = 3306
DATABASE_USER = "root"
DATABASE_PASSWORD = "OSp!jmgm@26"
DATABASE_NAME = "metrix"
DATABASE_TABLE = "interference_hourly_summary"
HEADER_FDD = (
"开始时间",
"粒度",
@@ -138,10 +146,6 @@ SUMMARY_HEADER = (
"source_path",
)
DEFAULT_DB_TABLE = "interference_hourly_summary"
DB_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9_]+$")
class ProcessingError(RuntimeError):
pass
@@ -165,24 +169,8 @@ class SummaryStore(Protocol):
class MySQLSummaryStore:
def __init__(
self,
host: str,
port: int,
user: str,
password: str,
database: str,
table: str = DEFAULT_DB_TABLE,
connect_factory: Callable[[], object] | None = None,
) -> None:
if not DB_IDENTIFIER_RE.fullmatch(table):
raise ProcessingError(f"Invalid database table name: {table!r}")
self.host = host
self.port = port
self.user = user
self.password = password
self.database = database
self.table = table
def __init__(self, connect_factory: Callable[[], object] | None = None) -> None:
self.table = DATABASE_TABLE
self.connect_factory = connect_factory
def replace_latest(self, rows: list[dict[str, str]]) -> dict[str, object]:
@@ -273,11 +261,11 @@ class MySQLSummaryStore:
raise ProcessingError("PyMySQL is required for database output") from exc
try:
return pymysql.connect(
host=self.host,
port=self.port,
user=self.user,
password=self.password,
database=self.database,
host=DATABASE_HOST,
port=DATABASE_PORT,
user=DATABASE_USER,
password=DATABASE_PASSWORD,
database=DATABASE_NAME,
charset="utf8mb4",
autocommit=False,
connect_timeout=15,
@@ -438,7 +426,6 @@ def process(
output_root: Path,
lookback_days: int,
requested_window: str = "",
plmn: str = "460-00",
store: SummaryStore | None = None,
) -> Path:
candidates = source.candidates(lookback_days)
@@ -463,7 +450,7 @@ def process(
write_csv(converted_dir / csv_name, header, rows)
for row_number, row in enumerate(rows, start=2):
record = dict(zip(header, row, strict=True))
summary_rows.append(summary_record(record, source_type, candidate.path, window, plmn, row_number))
summary_rows.append(summary_record(record, source_type, candidate.path, window, row_number))
manifest_files.append(
{
"source_type": source_type,
@@ -518,7 +505,7 @@ def process(
def summary_record(
record: dict[str, object], source_type: str, source_path: str, window: str, plmn: str, row_number: int
record: dict[str, object], source_type: str, source_path: str, window: str, row_number: int
) -> dict[str, str]:
hour_start = normalize_cell(record["开始时间"])
expected_start, hour_end = window_bounds(window)
@@ -526,19 +513,19 @@ def summary_record(
raise ProcessingError(f"{source_path}: row {row_number} time {hour_start!r} does not match {expected_start!r}")
if source_type in ("5G干扰监控", "700M干扰监控"):
cgi = required(record, "masterOperatorId", source_path, row_number)
cgi = build_cgi("", record, ("gNBplmn", "gNBId", "cellId"), source_path, row_number)
cell_name = required(record, "CU小区配置名称", source_path, row_number)
interference = required(record, "小区上行平均干扰电平(dBm)", source_path, row_number)
elif source_type in ("SDR_FDD干扰监控", "SDR_TDD干扰监控"):
cgi = build_cgi(plmn, record, "eNodeBID", "小区ID", source_path, row_number)
cgi = build_cgi(LTE_PLMN, record, ("eNodeBID", "小区ID"), source_path, row_number)
cell_name = required(record, "小区名称", source_path, row_number)
interference = required(record, "载波平均噪声干扰(dBm)", source_path, row_number)
elif source_type == "反开RD干扰监控":
cgi = build_cgi(plmn, record, "eNodeBId", "cellId", source_path, row_number)
cgi = build_cgi(LTE_PLMN, record, ("eNodeBId", "cellId"), source_path, row_number)
cell_name = required(record, "E-UTRAN TDD小区名称", source_path, row_number)
interference = required(record, "载波平均噪声干扰(dBm)", source_path, row_number)
else:
cgi = build_cgi(plmn, record, "eNodeBId", "cellId", source_path, row_number)
cgi = build_cgi(LTE_PLMN, record, ("eNodeBId", "cellId"), source_path, row_number)
cell_name = required(record, "E-UTRAN FDD小区名称", source_path, row_number)
interference = required(record, "载波平均噪声干扰(dBm)", source_path, row_number)
@@ -553,8 +540,9 @@ def summary_record(
}
def build_cgi(plmn: str, record: dict[str, object], node_column: str, cell_column: str, path: str, row: int) -> str:
return f"{plmn}-{required(record, node_column, path, row)}-{required(record, cell_column, path, row)}"
def build_cgi(prefix: str, record: dict[str, object], columns: tuple[str, ...], path: str, row: int) -> str:
parts = [required(record, column, path, row) for column in columns]
return "-".join(([prefix] if prefix else []) + parts)
def required(record: dict[str, object], column: str, path: str, row: int) -> str:
@@ -616,46 +604,17 @@ def build_parser() -> argparse.ArgumentParser:
parser.add_argument("--api-token", default=os.getenv("METRIX_API_TOKEN", ""))
parser.add_argument("--storage-id", default=os.getenv("METRIX_STORAGE_ID", DEFAULT_STORAGE_ID))
parser.add_argument("--root", default=os.getenv("INTERFERENCE_SOURCE_ROOT", DEFAULT_ROOT))
parser.add_argument("--plmn", default=os.getenv("INTERFERENCE_PLMN", "460-00"))
parser.add_argument("--no-database", action="store_true", help="Generate files without writing MySQL")
parser.add_argument("--db-host", default=os.getenv("INTERFERENCE_DB_HOST", ""))
parser.add_argument("--db-port", type=int, default=int(os.getenv("INTERFERENCE_DB_PORT", "3306")))
parser.add_argument("--db-user", default=os.getenv("INTERFERENCE_DB_USER", ""))
parser.add_argument("--db-password", default=os.getenv("INTERFERENCE_DB_PASSWORD", ""))
parser.add_argument("--db-name", default=os.getenv("INTERFERENCE_DB_NAME", ""))
parser.add_argument("--db-table", default=os.getenv("INTERFERENCE_DB_TABLE", DEFAULT_DB_TABLE))
return parser
def mysql_store_from_args(args: argparse.Namespace) -> MySQLSummaryStore:
required_values = {
"INTERFERENCE_DB_HOST": args.db_host,
"INTERFERENCE_DB_USER": args.db_user,
"INTERFERENCE_DB_PASSWORD": args.db_password,
"INTERFERENCE_DB_NAME": args.db_name,
}
missing = [name for name, value in required_values.items() if not value]
if missing:
raise ProcessingError(f"Missing database settings: {', '.join(missing)}; use --no-database for file-only runs")
if not 1 <= args.db_port <= 65535:
raise ProcessingError("db-port must be between 1 and 65535")
return MySQLSummaryStore(
host=args.db_host,
port=args.db_port,
user=args.db_user,
password=args.db_password,
database=args.db_name,
table=args.db_table,
)
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
if args.lookback_days < 1:
raise ProcessingError("lookback-days must be at least 1")
source: Source = LocalSource(args.source_dir) if args.source_dir else ApiSource(args.api_base, args.api_token, args.storage_id, args.root)
store = None if args.no_database else mysql_store_from_args(args)
process(source, args.output_dir, args.lookback_days, args.window, args.plmn, store)
store = None if args.no_database else MySQLSummaryStore()
process(source, args.output_dir, args.lookback_days, args.window, store=store)
return 0