File size: 2,530 Bytes
0ed457d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """Shared indexer singletons, locks, and 3GPP working-group/meeting-folder helpers.
Extracted from v0's app.py so that both the REST endpoints (used by the web
dashboard) and the MCP indexing server share the same indexer instances and
locks, preventing an indexing job started from one surface from racing with
one started from the other.
"""
import re
import threading
from classes import TDocIndexer, Spec3GPPIndexer, SpecETSIIndexer
meetings_mapping = {
"SA": [
"TSG_SA",
"WG1_Serv",
"WG2_Arch",
"WG3_Security",
"WG4_CODEC",
"WG5_TM",
"WG6_MissionCritical"
],
"CT": [
"TSG_CT",
"WG1_mm-cc-sm_ex-CN1",
"WG2_capability_ex-T2",
"WG3_interworking_ex-CN3",
"WG4_protocollars_ex-CN4",
"WG5_osa_ex-CN5",
"WG6_Smartcard_Ex-T3"
],
"RAN": [
"TSG_RAN",
"WG1_RL1",
"WG2_RL2",
"WG3_Iu",
"WG4_Radio",
"WG5_Test_ex-T1",
"WG6_legacyRAN"
]
}
def get_folder_name(working_group: str):
if working_group.endswith("P"):
if working_group.startswith("S"):
return ("SA", 0)
if working_group.startswith("C"):
return ("CT", 0)
if working_group.startswith("R"):
return ("RAN", 0)
m = re.match(r"([A-Z]+)(\d+)", working_group)
if m:
code, num = m.groups()
return (code, int(num))
else:
raise ValueError("Unattended format")
_tdoc_indexer = None
_spec_3gpp_indexer = None
_spec_etsi_indexer = None
_init_locks = {
"tdoc": threading.Lock(),
"3gpp": threading.Lock(),
"etsi": threading.Lock(),
}
indexing_locks = {
"tdoc": threading.Lock(),
"3gpp": threading.Lock(),
"etsi": threading.Lock(),
}
def get_tdoc_indexer():
global _tdoc_indexer
if _tdoc_indexer is None:
with _init_locks["tdoc"]:
if _tdoc_indexer is None:
_tdoc_indexer = TDocIndexer()
return _tdoc_indexer
def get_spec_3gpp_indexer():
global _spec_3gpp_indexer
if _spec_3gpp_indexer is None:
with _init_locks["3gpp"]:
if _spec_3gpp_indexer is None:
_spec_3gpp_indexer = Spec3GPPIndexer()
return _spec_3gpp_indexer
def get_spec_etsi_indexer():
global _spec_etsi_indexer
if _spec_etsi_indexer is None:
with _init_locks["etsi"]:
if _spec_etsi_indexer is None:
_spec_etsi_indexer = SpecETSIIndexer()
return _spec_etsi_indexer
|