PowerOn/platform-core
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

import cleanup
Some checks failed
Deploy Plattform-Core (Int) / test (push) Failing after 56s
Details
Deploy Plattform-Core (Int) / deploy (push) Has been skipped
Details

Browse source
This commit is contained in:
ValueOn AG 2026-06-11 22:54:39 +02:00
parent a1d9c68604
commit 63e30c1281
55 changed files with 1691 additions and 149 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
app.py
View file

@ -380,7 +380,6 @@ async def lifespan(app: FastAPI):
# Register all feature definitions in RBAC catalog (for /api/features/ endpoint) # Register all feature definitions in RBAC catalog (for /api/features/ endpoint)
try: try:
from modules.security.rbacCatalog import getCatalogService from modules.security.rbacCatalog import getCatalogService
from modules.system.registry import registerAllFeaturesInCatalog, syncCatalogFeaturesToDb
catalogService = getCatalogService() catalogService = getCatalogService()
registerAllFeaturesInCatalog(catalogService) registerAllFeaturesInCatalog(catalogService)
logger.info("Feature catalog registration completed") logger.info("Feature catalog registration completed")
@ -494,7 +493,6 @@ async def lifespan(app: FastAPI):
def _onRunFailed(workflowId, runId, error, mandateId=None, workflowLabel=None): def _onRunFailed(workflowId, runId, error, mandateId=None, workflowLabel=None):
from modules.serviceCenter import getService from modules.serviceCenter import getService
from modules.serviceCenter.context import ServiceCenterContext from modules.serviceCenter.context import ServiceCenterContext
from modules.interfaces.interfaceDbApp import getRootInterface
from modules.datamodels.datamodelMessaging import MessagingEventParameters from modules.datamodels.datamodelMessaging import MessagingEventParameters
rootInterface = getRootInterface() rootInterface = getRootInterface()
@ -555,6 +553,10 @@ async def lifespan(app: FastAPI):
from modules.serviceCenter.services.serviceSubscription.enterpriseRenewalScheduler import registerEnterpriseRenewalScheduler from modules.serviceCenter.services.serviceSubscription.enterpriseRenewalScheduler import registerEnterpriseRenewalScheduler
registerEnterpriseRenewalScheduler() registerEnterpriseRenewalScheduler()
# Register token and trusted device cleanup scheduler
from modules.auth.trustedDeviceService import registerTokenCleanupScheduler
registerTokenCleanupScheduler()
# Recover background jobs that were RUNNING when the previous worker died # Recover background jobs that were RUNNING when the previous worker died
try: try:
from modules.serviceCenter.services.serviceBackgroundJobs.mainBackgroundJobService import ( from modules.serviceCenter.services.serviceBackgroundJobs.mainBackgroundJobService import (
@ -901,6 +903,10 @@ app.include_router(demoConfigRouter)
from modules.routes.routeAdminDatabaseHealth import router as adminDatabaseHealthRouter from modules.routes.routeAdminDatabaseHealth import router as adminDatabaseHealthRouter
app.include_router(adminDatabaseHealthRouter) app.include_router(adminDatabaseHealthRouter)
from modules.routes.routeAdminSessions import router as adminSessionsRouter, trustedDeviceRouter as adminTrustedDeviceRouter
app.include_router(adminSessionsRouter)
app.include_router(adminTrustedDeviceRouter)
from modules.routes.routeGdpr import router as gdprRouter from modules.routes.routeGdpr import router as gdprRouter
app.include_router(gdprRouter) app.include_router(gdprRouter)

2
env-dev.env
View file

@ -58,7 +58,7 @@ Service_CLICKUP_OAUTH_REDIRECT_URI = http://localhost:8000/api/clickup/auth/conn
# Stripe Billing (both end with _SECRET for encryption script) # Stripe Billing (both end with _SECRET for encryption script)
STRIPE_SECRET_KEY_SECRET = DEV_ENC:Z0FBQUFBQnB5dkd5aHNGejgzQmpTdmprdzQxR19KZkh3MlhYUTNseFN3WnlaWjh2SDZyalN6aU9xSktkbUQwUnZrVnlvbGVRQm4yZFdiRU5aSEk5WVJuUnR4VUwtTm9OVk1WWmJQeU5QaDdib0hfVWV5U1BfYTFXRmdoOWdnOWxkb3JFQmF3bm45UjFUVUxmWGtGRkFKUGd6bmhpQlFnaVI3Q2lLdDlsY1VESk1vOEM0ZFBJNW1qcVZ0N2tPYmRLNmVKajZ2M3o3S05lWnRRVG5LdkRseW4wQ3VjNHNQZTZUdz09 STRIPE_SECRET_KEY_SECRET = DEV_ENC:Z0FBQUFBQnB5dkd5aHNGejgzQmpTdmprdzQxR19KZkh3MlhYUTNseFN3WnlaWjh2SDZyalN6aU9xSktkbUQwUnZrVnlvbGVRQm4yZFdiRU5aSEk5WVJuUnR4VUwtTm9OVk1WWmJQeU5QaDdib0hfVWV5U1BfYTFXRmdoOWdnOWxkb3JFQmF3bm45UjFUVUxmWGtGRkFKUGd6bmhpQlFnaVI3Q2lLdDlsY1VESk1vOEM0ZFBJNW1qcVZ0N2tPYmRLNmVKajZ2M3o3S05lWnRRVG5LdkRseW4wQ3VjNHNQZTZUdz09
STRIPE_WEBHOOK_SECRET = DEV_ENC:Z0FBQUFBQnB5dkd5dDJMSHBrVk8wTzJhU2xzTTZCZWdvWmU2NGI2WklfRXRJZVUzaVYyOU9GLUZsalUwa2lPdEgtUHo0dVVvRDU1cy1saHJyU0Rxa2xQZjBuakExQzk3bmxBcU9WbEIxUEtpR1JoUFMxZG9ISGRZUXFhdFpSMGxvQUV3a0VLQllfUUtCOHZwTGdteV9rYTFOazBfSlN3ekNWblFpakJlZVlCTmNkWWQ4Sm01a1RCWTlnTlFHWVA0MkZYMlprUExrWFN2V0NVU1BTd1NKczFJbVo3VHpLdlc4UT09 STRIPE_WEBHOOK_SECRET = DEV_ENC:Z0FBQUFBQnFLeUFlb2dfSjZPaWIyRjZsNjhiSDFQNFpxdW50YmlLUjFLX1lJMGdCWUtBUEdrRGhvSzVVWnkxNVZEdmtkQmk5X05YS0JVU1NyX3VQZTV2VjVwakd0RGM2WUl6TTlzbms1d1NCOTQtdURiVjhxdXZGVlR1ZVNTbUkwOFh1R04yUUxxay0=
STRIPE_API_VERSION = 2026-01-28.clover STRIPE_API_VERSION = 2026-01-28.clover
STRIPE_AUTOMATIC_TAX_ENABLED = false STRIPE_AUTOMATIC_TAX_ENABLED = false
STRIPE_TAX_RATE_ID_CH_VAT = txr_1TOQd14OUoIL0Osj7A0ZQlr0 STRIPE_TAX_RATE_ID_CH_VAT = txr_1TOQd14OUoIL0Osj7A0ZQlr0

2
env-int.env
View file

@ -60,7 +60,7 @@ Service_CLICKUP_OAUTH_REDIRECT_URI = https://api-int.poweron.swiss/api/clickup/a
# Stripe Billing (both end with _SECRET for encryption script) # Stripe Billing (both end with _SECRET for encryption script)
STRIPE_SECRET_KEY_SECRET = INT_ENC:Z0FBQUFBQnB5dkd5ekdBaGNGVUlOQUpncTlzLWlTV0V5OWZzQkpDczhCUGw4U1JpTHZ0d3pfYlFNWElLRlNiNlNsaDRYTGZUTkg2OUFrTW1GZXpOUjBVbmRQWjN6ekhHd2ZSQ195OHlaeWh1TmxrUm10V2R3YmdncmFLbFMzVjdqcWJMSUJPR2xuSEozclNoZG1rZVBTaWg3OFQ1Qzdxb0wyQ2RKazc2dG1aZXBUTXlvbDZqLS1KOVI5M3BGc3NQZkZRbnFpRjIwWmh2ZHlVNlpxZVo2dWNmMjQ5eW02QmtzUT09 STRIPE_SECRET_KEY_SECRET = INT_ENC:Z0FBQUFBQnB5dkd5ekdBaGNGVUlOQUpncTlzLWlTV0V5OWZzQkpDczhCUGw4U1JpTHZ0d3pfYlFNWElLRlNiNlNsaDRYTGZUTkg2OUFrTW1GZXpOUjBVbmRQWjN6ekhHd2ZSQ195OHlaeWh1TmxrUm10V2R3YmdncmFLbFMzVjdqcWJMSUJPR2xuSEozclNoZG1rZVBTaWg3OFQ1Qzdxb0wyQ2RKazc2dG1aZXBUTXlvbDZqLS1KOVI5M3BGc3NQZkZRbnFpRjIwWmh2ZHlVNlpxZVo2dWNmMjQ5eW02QmtzUT09
STRIPE_WEBHOOK_SECRET = INT_ENC:Z0FBQUFBQnA4UXZiUUVqTl9lREVRWTh1aHFDcFpwcXRkOUx4MS1ham9Ddkl6T0xzMnJuM1hhUHdGNG5CenY1MUg4RlJBOGFQTWl5cVd5MjJ2REItcHYyRmdLX3ZlT2p5Z3BRVkMtQnRoTVkteXlfaU92MVBtOEI0Ni1kbGlfa0NiRmFRRXNHLVE2NHI= STRIPE_WEBHOOK_SECRET = INT_ENC:Z0FBQUFBQnFLeUFWUUtMZ25NQ2ZOWE5nRF9CaFNwcXhSU2tKRktLaElLRHJMM295OXNkVEFLekVUMzN0YUpIZHJfWGNqa0xxOFZRVHZEUXVLZ3ItVGZWc2VFQ2thcUlJalY1b0JDSmR6RF96d1A3OGhyd0w1MHZPeFNZRkl0c19kYUJQcHVwR2tsd0s=
STRIPE_API_VERSION = 2026-01-28.clover STRIPE_API_VERSION = 2026-01-28.clover
STRIPE_AUTOMATIC_TAX_ENABLED = false STRIPE_AUTOMATIC_TAX_ENABLED = false
STRIPE_TAX_RATE_ID_CH_VAT = txr_1TOQd14OUoIL0Osj7A0ZQlr0 STRIPE_TAX_RATE_ID_CH_VAT = txr_1TOQd14OUoIL0Osj7A0ZQlr0

2
env-prod.env
View file

@ -58,7 +58,7 @@ Service_CLICKUP_OAUTH_REDIRECT_URI = https://api.poweron.swiss/api/clickup/auth/
# Stripe Billing (both end with _SECRET for encryption script) # Stripe Billing (both end with _SECRET for encryption script)
STRIPE_SECRET_KEY_SECRET = PROD_ENC:Z0FBQUFBQnB5dkd6aVA3R3VRS3VHMUgzUEVjYkR4eUZKWFhPUzFTTVlHNnBvT3FienNQaUlBWVpPLXJyVGpGMWk4LXktMXphX0J6ZTVESkJxdjNNa3ZJbF9wX2ppYzdjYlF0cmdVamlEWWJDSmJYYkJseHctTlh4dnNoQWs4SG5haVl2TTNDdXpuaFpqeDBtNkFCbUxMa0RaWG14dmxyOEdILTNrZ2licmNpbXVkN2lFSWoxZW1BODNpV0ZTQ0VaeXRmR1d4RjExMlVFS3MtQU9zZXZlZE1mTmY3OWctUXJHdz09 STRIPE_SECRET_KEY_SECRET = PROD_ENC:Z0FBQUFBQnB5dkd6aVA3R3VRS3VHMUgzUEVjYkR4eUZKWFhPUzFTTVlHNnBvT3FienNQaUlBWVpPLXJyVGpGMWk4LXktMXphX0J6ZTVESkJxdjNNa3ZJbF9wX2ppYzdjYlF0cmdVamlEWWJDSmJYYkJseHctTlh4dnNoQWs4SG5haVl2TTNDdXpuaFpqeDBtNkFCbUxMa0RaWG14dmxyOEdILTNrZ2licmNpbXVkN2lFSWoxZW1BODNpV0ZTQ0VaeXRmR1d4RjExMlVFS3MtQU9zZXZlZE1mTmY3OWctUXJHdz09
STRIPE_WEBHOOK_SECRET = PROD_ENC:Z0FBQUFBQnBudkpGNUpTWldsakYydFhFelBrR1lSaWxYT3kyMENOMUljZTJUZHBWcEhhdWVCMzYxZXQ5b3VlTFVRalFiTVdsbGxrdUx0RDFwSEpsOC1sTDJRTEJNQlA3S3ZaQzBtV1h6bWp5VnlMZUgwUlF3cXYxcnljZVE5SWdzLVg3V0syOWRYS08= STRIPE_WEBHOOK_SECRET = PROD_ENC:Z0FBQUFBQnFLeUFNQ1FhVE94ZzM3V3NCVGVVWnltUndsOG1Ra0hQTmJ3QWY5aXVWeTJsX3A4a3VBSnFQd3drWFRZNFVDdWxCeFgyQ0RpNGQ0SlJOcm9tVE5KZmVqQU1WUjFjeDRJeGE5THdmR0g1V2dQUk5SSjcySnAzR245NW5NUFVDT3lJUWpjWFo=
STRIPE_API_VERSION = 2026-01-28.clover STRIPE_API_VERSION = 2026-01-28.clover
STRIPE_AUTOMATIC_TAX_ENABLED = false STRIPE_AUTOMATIC_TAX_ENABLED = false
STRIPE_TAX_RATE_ID_CH_VAT = txr_1TOQZG8WqlVsabrfFEu49pah STRIPE_TAX_RATE_ID_CH_VAT = txr_1TOQZG8WqlVsabrfFEu49pah

48
modules/auth/trustedDeviceService.py
View file

@ -169,3 +169,51 @@ def _getClientIp(request: Request) -> Optional[str]:
if request.client: if request.client:
return request.client.host return request.client.host
return None return None
# --- Scheduler Integration ---
async def _runTokenAndDeviceCleanup() -> None:
"""Scheduled task: remove expired tokens and trusted devices."""
try:
from modules.connectors.connectorDbPostgre import ConnectorPostgre
db = ConnectorPostgre("poweron_app")
now = getUtcTimestamp()
# Expired auth-session tokens
tokens = db.getRecordset(
Token,
recordFilter={"tokenPurpose": TokenPurpose.AUTH_SESSION.value},
)
expiredCount = 0
for t in tokens:
if t.get("expiresAt", 0) < now:
db.recordDelete(Token, t["id"])
expiredCount += 1
# Expired trusted devices
deviceCount = cleanupExpiredDevices(db)
if expiredCount or deviceCount:
logger.info(
f"Token cleanup: {expiredCount} expired token(s), "
f"{deviceCount} expired trusted device(s) removed"
)
except Exception as e:
logger.error(f"Token/device cleanup failed: {e}")
def registerTokenCleanupScheduler() -> None:
"""Register daily token cleanup job. Call during app startup."""
try:
from modules.shared.eventManagement import eventManager
eventManager.registerCron(
jobId="token_device_cleanup",
func=_runTokenAndDeviceCleanup,
cronKwargs={"hour": "4", "minute": "0"},
)
logger.info("Token/device cleanup scheduler registered (daily 04:00)")
except Exception as e:
logger.warning(f"Failed to register token cleanup scheduler: {e}")

11
modules/features/commcoach/interfaceFeatureCommcoach.py
View file

@ -261,35 +261,29 @@ class CommcoachObjects:
# ========================================================================= # =========================================================================
def getPersonas(self, userId: str, instanceId: str) -> List[Dict[str, Any]]: def getPersonas(self, userId: str, instanceId: str) -> List[Dict[str, Any]]:
from .datamodelCommcoach import CoachingPersona
builtins = self.db.getRecordset(CoachingPersona, recordFilter={"userId": "system"}) builtins = self.db.getRecordset(CoachingPersona, recordFilter={"userId": "system"})
custom = self.db.getRecordset(CoachingPersona, recordFilter={"userId": userId, "instanceId": instanceId}) custom = self.db.getRecordset(CoachingPersona, recordFilter={"userId": userId, "instanceId": instanceId})
all = builtins + custom all = builtins + custom
return [p for p in all if p.get("isActive", True)] return [p for p in all if p.get("isActive", True)]
def getPersona(self, personaId: str) -> Optional[Dict[str, Any]]: def getPersona(self, personaId: str) -> Optional[Dict[str, Any]]:
from .datamodelCommcoach import CoachingPersona
records = self.db.getRecordset(CoachingPersona, recordFilter={"id": personaId}) records = self.db.getRecordset(CoachingPersona, recordFilter={"id": personaId})
return records[0] if records else None return records[0] if records else None
def createPersona(self, data: Dict[str, Any]) -> Dict[str, Any]: def createPersona(self, data: Dict[str, Any]) -> Dict[str, Any]:
from .datamodelCommcoach import CoachingPersona
data["createdAt"] = getIsoTimestamp() data["createdAt"] = getIsoTimestamp()
data["updatedAt"] = getIsoTimestamp() data["updatedAt"] = getIsoTimestamp()
return self.db.recordCreate(CoachingPersona, data) return self.db.recordCreate(CoachingPersona, data)
def updatePersona(self, personaId: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]: def updatePersona(self, personaId: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]:
from .datamodelCommcoach import CoachingPersona
updates["updatedAt"] = getIsoTimestamp() updates["updatedAt"] = getIsoTimestamp()
return self.db.recordModify(CoachingPersona, personaId, updates) return self.db.recordModify(CoachingPersona, personaId, updates)
def deletePersona(self, personaId: str) -> bool: def deletePersona(self, personaId: str) -> bool:
from .datamodelCommcoach import CoachingPersona
return self.db.recordDelete(CoachingPersona, personaId) return self.db.recordDelete(CoachingPersona, personaId)
def getAllPersonas(self, instanceId: str) -> List[Dict[str, Any]]: def getAllPersonas(self, instanceId: str) -> List[Dict[str, Any]]:
"""All personas (builtin + custom for this instance), including inactive.""" """All personas (builtin + custom for this instance), including inactive."""
from .datamodelCommcoach import CoachingPersona
builtins = self.db.getRecordset(CoachingPersona, recordFilter={"userId": "system"}) builtins = self.db.getRecordset(CoachingPersona, recordFilter={"userId": "system"})
custom = self.db.getRecordset(CoachingPersona, recordFilter={"instanceId": instanceId}) custom = self.db.getRecordset(CoachingPersona, recordFilter={"instanceId": instanceId})
custom = [p for p in custom if p.get("userId") != "system"] custom = [p for p in custom if p.get("userId") != "system"]
@ -300,11 +294,9 @@ class CommcoachObjects:
# ========================================================================= # =========================================================================
def getModulePersonas(self, moduleId: str) -> List[Dict[str, Any]]: def getModulePersonas(self, moduleId: str) -> List[Dict[str, Any]]:
from .datamodelCommcoach import ModulePersonaMapping
return self.db.getRecordset(ModulePersonaMapping, recordFilter={"moduleId": moduleId}) return self.db.getRecordset(ModulePersonaMapping, recordFilter={"moduleId": moduleId})
def setModulePersonas(self, moduleId: str, personaIds: List[str], instanceId: str) -> List[Dict[str, Any]]: def setModulePersonas(self, moduleId: str, personaIds: List[str], instanceId: str) -> List[Dict[str, Any]]:
from .datamodelCommcoach import ModulePersonaMapping
existing = self.db.getRecordset(ModulePersonaMapping, recordFilter={"moduleId": moduleId}) existing = self.db.getRecordset(ModulePersonaMapping, recordFilter={"moduleId": moduleId})
for rec in existing: for rec in existing:
self.db.recordDelete(ModulePersonaMapping, rec["id"]) self.db.recordDelete(ModulePersonaMapping, rec["id"])
@ -325,18 +317,15 @@ class CommcoachObjects:
# ========================================================================= # =========================================================================
def getBadges(self, userId: str, instanceId: str) -> List[Dict[str, Any]]: def getBadges(self, userId: str, instanceId: str) -> List[Dict[str, Any]]:
from .datamodelCommcoach import CoachingBadge
records = self.db.getRecordset(CoachingBadge, recordFilter={"userId": userId, "instanceId": instanceId}) records = self.db.getRecordset(CoachingBadge, recordFilter={"userId": userId, "instanceId": instanceId})
records.sort(key=lambda r: r.get("awardedAt") or 0, reverse=True) records.sort(key=lambda r: r.get("awardedAt") or 0, reverse=True)
return records return records
def hasBadge(self, userId: str, instanceId: str, badgeKey: str) -> bool: def hasBadge(self, userId: str, instanceId: str, badgeKey: str) -> bool:
from .datamodelCommcoach import CoachingBadge
records = self.db.getRecordset(CoachingBadge, recordFilter={"userId": userId, "instanceId": instanceId, "badgeKey": badgeKey}) records = self.db.getRecordset(CoachingBadge, recordFilter={"userId": userId, "instanceId": instanceId, "badgeKey": badgeKey})
return len(records) > 0 return len(records) > 0
def awardBadge(self, data: Dict[str, Any]) -> Dict[str, Any]: def awardBadge(self, data: Dict[str, Any]) -> Dict[str, Any]:
from .datamodelCommcoach import CoachingBadge
data["awardedAt"] = getUtcTimestamp() data["awardedAt"] = getUtcTimestamp()
data["createdAt"] = getIsoTimestamp() data["createdAt"] = getIsoTimestamp()
return self.db.recordCreate(CoachingBadge, data) return self.db.recordCreate(CoachingBadge, data)

5
modules/features/commcoach/routeFeatureCommcoach.py
View file

@ -333,7 +333,6 @@ async def startSession(
try: try:
from modules.interfaces.interfaceVoiceObjects import getVoiceInterface from modules.interfaces.interfaceVoiceObjects import getVoiceInterface
voiceInterface = getVoiceInterface(context.user, mandateId) voiceInterface = getVoiceInterface(context.user, mandateId)
from .serviceCommcoach import getUserVoicePrefs, stripMarkdownForTts, buildTtsConfigErrorMessage
language, voiceName = getUserVoicePrefs(userId, mandateId) language, voiceName = getUserVoicePrefs(userId, mandateId)
ttsResult = await voiceInterface.textToSpeech( ttsResult = await voiceInterface.textToSpeech(
text=stripMarkdownForTts(greetingText), text=stripMarkdownForTts(greetingText),
@ -378,7 +377,6 @@ async def startSession(
asyncio.create_task(service.processSessionOpening(sessionId, moduleId, interface)) asyncio.create_task(service.processSessionOpening(sessionId, moduleId, interface))
async def _newSessionEventGenerator(): async def _newSessionEventGenerator():
from modules.shared.timeUtils import getIsoTimestamp
timeoutCount = 0 timeoutCount = 0
try: try:
while True: while True:
@ -468,7 +466,6 @@ async def cancelSession(
raise HTTPException(status_code=404, detail=routeApiMsg("Session not found")) raise HTTPException(status_code=404, detail=routeApiMsg("Session not found"))
_validateOwnership(session, context) _validateOwnership(session, context)
from modules.shared.timeUtils import getUtcTimestamp
interface.updateSession(sessionId, { interface.updateSession(sessionId, {
"status": CoachingSessionStatus.CANCELLED.value, "status": CoachingSessionStatus.CANCELLED.value,
"endedAt": getUtcTimestamp(), "endedAt": getUtcTimestamp(),
@ -581,7 +578,6 @@ async def sendAudioStream(
if not audioBody: if not audioBody:
raise HTTPException(status_code=400, detail=routeApiMsg("No audio data received")) raise HTTPException(status_code=400, detail=routeApiMsg("No audio data received"))
from .serviceCommcoach import getUserVoicePrefs
language, _ = getUserVoicePrefs(str(context.user.id), mandateId) language, _ = getUserVoicePrefs(str(context.user.id), mandateId)
moduleId = session.get("moduleId") moduleId = session.get("moduleId")
@ -765,7 +761,6 @@ async def updateTaskStatus(
updates = {"status": body.status.value} updates = {"status": body.status.value}
if body.status == CoachingTaskStatus.DONE: if body.status == CoachingTaskStatus.DONE:
from modules.shared.timeUtils import getUtcTimestamp
updates["completedAt"] = getUtcTimestamp() updates["completedAt"] = getUtcTimestamp()
updated = interface.updateTask(taskId, updates) updated = interface.updateTask(taskId, updates)

4
modules/features/commcoach/serviceCommcoach.py
View file

@ -98,7 +98,6 @@ def getUserVoicePrefs(userId: str, mandateId: Optional[str] = None) -> tuple:
"""Load voice language and voiceName from central UserVoicePreferences. """Load voice language and voiceName from central UserVoicePreferences.
Returns (language, voiceName) tuple.""" Returns (language, voiceName) tuple."""
try: try:
from modules.datamodels.datamodelUam import UserVoicePreferences
from modules.interfaces.interfaceDbApp import getRootInterface from modules.interfaces.interfaceDbApp import getRootInterface
rootIf = getRootInterface() rootIf = getRootInterface()
prefs = rootIf.db.getRecordset( prefs = rootIf.db.getRecordset(
@ -430,7 +429,6 @@ async def _resolveDocumentIntent(combinedUserPrompt: str, docs: List[Dict[str, A
"""Pre-AI-call: identify which documents the user references and what action is needed.""" """Pre-AI-call: identify which documents the user references and what action is needed."""
if not docs: if not docs:
return {"read": [], "update": [], "create": [], "noDocumentAction": True} return {"read": [], "update": [], "create": [], "noDocumentAction": True}
from . import serviceCommcoachAi as aiPrompts
docCatalog = [{"id": d.get("id", ""), "title": d.get("summary") or d.get("fileName", ""), "summary": (d.get("summary") or "")[:100]} for d in docs] docCatalog = [{"id": d.get("id", ""), "title": d.get("summary") or d.get("fileName", ""), "summary": (d.get("summary") or "")[:100]} for d in docs]
prompt = aiPrompts.buildDocumentIntentPrompt(combinedUserPrompt, docCatalog) prompt = aiPrompts.buildDocumentIntentPrompt(combinedUserPrompt, docCatalog)
try: try:
@ -744,7 +742,6 @@ class CommcoachService:
4. Map agent events to CommCoach SSE events 4. Map agent events to CommCoach SSE events
5. Post-processing: store message, TTS, tasks, scores 5. Post-processing: store message, TTS, tasks, scores
""" """
from . import interfaceFeatureCommcoach as interfaceDb
# Store user message # Store user message
userMsg = CoachingMessage( userMsg = CoachingMessage(
@ -907,7 +904,6 @@ class CommcoachService:
) )
agentService = getService("agent", serviceContext) agentService = getService("agent", serviceContext)
from modules.datamodels.datamodelAi import PriorityEnum, OperationTypeEnum
config = AgentConfig( config = AgentConfig(
toolSet="commcoach" if useTools else "none", toolSet="commcoach" if useTools else "none",
maxRounds=3 if useTools else 1, maxRounds=3 if useTools else 1,

2
modules/features/neutralization/serviceNeutralization/subProcessList.py
View file

@ -157,7 +157,6 @@ class ListProcessor:
processedAttrs[attrName] = self.string_parser.mapping[attrValue] processedAttrs[attrName] = self.string_parser.mapping[attrValue]
else: else:
# Check if attribute value matches any data patterns # Check if attribute value matches any data patterns
from .subPatterns import findPatternsInText, DataPatterns
matches = findPatternsInText(attrValue, DataPatterns.patterns) matches = findPatternsInText(attrValue, DataPatterns.patterns)
if matches: if matches:
patternName = matches[0][0] patternName = matches[0][0]
@ -191,7 +190,6 @@ class ListProcessor:
# Skip if already a placeholder # Skip if already a placeholder
if not self.string_parser._isPlaceholder(text): if not self.string_parser._isPlaceholder(text):
# Check if text matches any patterns # Check if text matches any patterns
from .subPatterns import findPatternsInText, DataPatterns
patternMatches = findPatternsInText(text, DataPatterns.patterns) patternMatches = findPatternsInText(text, DataPatterns.patterns)
if patternMatches: if patternMatches:

1
modules/features/realEstate/interfaceFeatureRealEstate.py
View file

@ -796,7 +796,6 @@ class RealEstateObjects:
return False return False
tableName = modelClass.__name__ tableName = modelClass.__name__
from modules.interfaces.interfaceRbac import buildDataObjectKey
objectKey = buildDataObjectKey(tableName, featureCode=self.featureCode if hasattr(self, 'featureCode') else None) objectKey = buildDataObjectKey(tableName, featureCode=self.featureCode if hasattr(self, 'featureCode') else None)
permissions = self.rbac.getUserPermissions( permissions = self.rbac.getUserPermissions(
self.currentUser, self.currentUser,

9
modules/features/teamsbot/routeFeatureTeamsbot.py
View file

@ -328,7 +328,6 @@ async def startSession(
if context.isSysAdmin and joinMode == TeamsbotJoinMode.SYSTEM_BOT: if context.isSysAdmin and joinMode == TeamsbotJoinMode.SYSTEM_BOT:
systemBot = interface.getActiveSystemBot(mandateId) systemBot = interface.getActiveSystemBot(mandateId)
if not systemBot: if not systemBot:
from .datamodelTeamsbot import TeamsbotSystemBot
allBots = interface.db.getRecordset(TeamsbotSystemBot, recordFilter={"isActive": True}) allBots = interface.db.getRecordset(TeamsbotSystemBot, recordFilter={"isActive": True})
if allBots: if allBots:
systemBot = allBots[0] systemBot = allBots[0]
@ -537,7 +536,6 @@ async def streamSession(
async def _eventGenerator(): async def _eventGenerator():
"""Generate SSE events from the session event queue.""" """Generate SSE events from the session event queue."""
from .service import sessionEvents
# Send initial session state with stats # Send initial session state with stats
stats = interface.getSessionStats(sessionId) stats = interface.getSessionStats(sessionId)
@ -545,7 +543,6 @@ async def streamSession(
# Send current bot WebSocket connection state so the operator UI can # Send current bot WebSocket connection state so the operator UI can
# render the live indicator without waiting for the next connect/disconnect. # render the live indicator without waiting for the next connect/disconnect.
from .service import getActiveService as _getActiveService
yield f"data: {json.dumps({'type': 'botConnectionState', 'data': {'connected': _getActiveService(sessionId) is not None}})}\n\n" yield f"data: {json.dumps({'type': 'botConnectionState', 'data': {'connected': _getActiveService(sessionId) is not None}})}\n\n"
# Stream events # Stream events
@ -1040,7 +1037,6 @@ async def submitDirectorPrompt(
detail=routeApiMsg(f"Too many files ({len(fileIds)}); max {DIRECTOR_PROMPT_FILE_LIMIT}"), detail=routeApiMsg(f"Too many files ({len(fileIds)}); max {DIRECTOR_PROMPT_FILE_LIMIT}"),
) )
from .service import getActiveService
service = getActiveService(sessionId) service = getActiveService(sessionId)
if not service: if not service:
raise HTTPException( raise HTTPException(
@ -1108,7 +1104,6 @@ async def deleteDirectorPrompt(
if not context.isPlatformAdmin and prompt.get("operatorUserId") != str(context.user.id): if not context.isPlatformAdmin and prompt.get("operatorUserId") != str(context.user.id):
raise HTTPException(status_code=404, detail=f"Prompt '{promptId}' not found") raise HTTPException(status_code=404, detail=f"Prompt '{promptId}' not found")
from .service import getActiveService
service = getActiveService(sessionId) service = getActiveService(sessionId)
if service: if service:
await service.removePersistentPrompt(promptId) await service.removePersistentPrompt(promptId)
@ -1134,7 +1129,6 @@ async def testVoice(
): ):
"""Test TTS voice with AI-generated sample text in the correct language.""" """Test TTS voice with AI-generated sample text in the correct language."""
from modules.interfaces.interfaceVoiceObjects import getVoiceInterface from modules.interfaces.interfaceVoiceObjects import getVoiceInterface
from .service import createAiService
from modules.datamodels.datamodelAi import AiCallRequest, AiCallOptions, OperationTypeEnum, PriorityEnum from modules.datamodels.datamodelAi import AiCallRequest, AiCallOptions, OperationTypeEnum, PriorityEnum
mandateId = _validateInstanceAccess(instanceId, context) mandateId = _validateInstanceAccess(instanceId, context)
@ -1547,7 +1541,6 @@ async def postTranscript(
originalUser = rootUser originalUser = rootUser
# Process transcript through the service pipeline # Process transcript through the service pipeline
from .service import TeamsbotService
from modules.interfaces.interfaceVoiceObjects import getVoiceInterface from modules.interfaces.interfaceVoiceObjects import getVoiceInterface
service = TeamsbotService(originalUser, mandateId, instanceId, config) service = TeamsbotService(originalUser, mandateId, instanceId, config)
@ -1600,7 +1593,6 @@ async def postBotStatus(
if not originalUser: if not originalUser:
originalUser = rootUser originalUser = rootUser
from .service import TeamsbotService
service = TeamsbotService(originalUser, mandateId, instanceId, config) service = TeamsbotService(originalUser, mandateId, instanceId, config)
interface = interfaceDb.getInterface(originalUser, mandateId=mandateId, featureInstanceId=instanceId) interface = interfaceDb.getInterface(originalUser, mandateId=mandateId, featureInstanceId=instanceId)
@ -1640,7 +1632,6 @@ async def botWebsocket(
# Load the original user who started the session (has RBAC roles in mandate) # Load the original user who started the session (has RBAC roles in mandate)
# Bot callbacks have no HTTP auth, so we reconstruct the user context from the session record. # Bot callbacks have no HTTP auth, so we reconstruct the user context from the session record.
from modules.interfaces.interfaceDbApp import getRootInterface
rootInterface = getRootInterface() rootInterface = getRootInterface()
rootUser = rootInterface.currentUser rootUser = rootInterface.currentUser

4
modules/features/trustee/accounting/accountingBridge.py
View file

@ -33,7 +33,6 @@ class AccountingBridge:
async def getActiveConfig(self, featureInstanceId: str) -> Optional[Dict[str, Any]]: async def getActiveConfig(self, featureInstanceId: str) -> Optional[Dict[str, Any]]:
"""Load the active TrusteeAccountingConfig for a feature instance.""" """Load the active TrusteeAccountingConfig for a feature instance."""
from modules.features.trustee.datamodelFeatureTrustee import TrusteeAccountingConfig
records = self._trusteeInterface.db.getRecordset( records = self._trusteeInterface.db.getRecordset(
TrusteeAccountingConfig, TrusteeAccountingConfig,
recordFilter={"featureInstanceId": featureInstanceId, "isActive": True}, recordFilter={"featureInstanceId": featureInstanceId, "isActive": True},
@ -128,7 +127,6 @@ class AccountingBridge:
Optional _resolved* params allow pushBatchToAccounting to pass a pre-resolved Optional _resolved* params allow pushBatchToAccounting to pass a pre-resolved
connector/config so we don't decrypt per position (avoids rate-limit). connector/config so we don't decrypt per position (avoids rate-limit).
""" """
from modules.features.trustee.datamodelFeatureTrustee import TrusteePosition, TrusteeAccountingSync
connector = _resolvedConnector connector = _resolvedConnector
plainConfig = _resolvedPlainConfig plainConfig = _resolvedPlainConfig
@ -306,7 +304,6 @@ class AccountingBridge:
# Update last sync on config record # Update last sync on config record
if configRecord: if configRecord:
from modules.features.trustee.datamodelFeatureTrustee import TrusteeAccountingConfig
updatePayload = { updatePayload = {
"lastSyncAt": time.time(), "lastSyncAt": time.time(),
"lastSyncStatus": "success" if result.success else "error", "lastSyncStatus": "success" if result.success else "error",
@ -335,7 +332,6 @@ class AccountingBridge:
async def refreshChartOfAccounts(self, featureInstanceId: str) -> List[AccountingChart]: async def refreshChartOfAccounts(self, featureInstanceId: str) -> List[AccountingChart]:
"""Fetch the full chart of accounts from the external system and cache it locally on TrusteeAccountingConfig.""" """Fetch the full chart of accounts from the external system and cache it locally on TrusteeAccountingConfig."""
from modules.features.trustee.datamodelFeatureTrustee import TrusteeAccountingConfig
connector, plainConfig, configRecord = await self._resolveConnectorAndConfig(featureInstanceId) connector, plainConfig, configRecord = await self._resolveConnectorAndConfig(featureInstanceId)
if not connector or not plainConfig or not configRecord: if not connector or not plainConfig or not configRecord:

2
modules/features/trustee/interfaceFeatureTrustee.py
View file

@ -309,7 +309,6 @@ class TrusteeObjects:
return False return False
tableName = modelClass.__name__ tableName = modelClass.__name__
from modules.interfaces.interfaceRbac import buildDataObjectKey
objectKey = buildDataObjectKey(tableName, featureCode=self.featureCode if hasattr(self, 'featureCode') else None) objectKey = buildDataObjectKey(tableName, featureCode=self.featureCode if hasattr(self, 'featureCode') else None)
permissions = self.rbac.getUserPermissions( permissions = self.rbac.getUserPermissions(
self.currentUser, self.currentUser,
@ -338,7 +337,6 @@ class TrusteeObjects:
return AccessLevel.NONE return AccessLevel.NONE
tableName = modelClass.__name__ tableName = modelClass.__name__
from modules.interfaces.interfaceRbac import buildDataObjectKey
objectKey = buildDataObjectKey(tableName, featureCode=self.featureCode if hasattr(self, 'featureCode') else None) objectKey = buildDataObjectKey(tableName, featureCode=self.featureCode if hasattr(self, 'featureCode') else None)
permissions = self.rbac.getUserPermissions( permissions = self.rbac.getUserPermissions(
self.currentUser, self.currentUser,

7
modules/features/trustee/routeFeatureTrustee.py
View file

@ -170,7 +170,6 @@ def getQuickActions(
if role and role.roleLabel: if role and role.roleLabel:
userRoleLabels.add(role.roleLabel) userRoleLabels.add(role.roleLabel)
from modules.shared.i18nRegistry import resolveText
lang = (language or "de").strip() or "de" lang = (language or "de").strip() or "de"
@ -1201,7 +1200,6 @@ def _buildSyncStatusByPosition(interface, instanceId: str) -> Dict[str, Dict[str
``error``, so a successful retry hides an old failure. Any other status ``error``, so a successful retry hides an old failure. Any other status
(`pending`, `cancelled`, ...) is kept verbatim. (`pending`, `cancelled`, ...) is kept verbatim.
""" """
from .datamodelFeatureTrustee import TrusteeAccountingSync
syncRecords = interface.db.getRecordset( syncRecords = interface.db.getRecordset(
TrusteeAccountingSync, recordFilter={"featureInstanceId": instanceId} TrusteeAccountingSync, recordFilter={"featureInstanceId": instanceId}
@ -1290,7 +1288,6 @@ def _handlePositionMode(instanceId, mandateId, mode, column, pagination, context
"""Handle mode=filterValues and mode=ids for trustee positions.""" """Handle mode=filterValues and mode=ids for trustee positions."""
from modules.dbHelpers.paginationHelpers import handleIdsInMemory, handleFilterValuesInMemory from modules.dbHelpers.paginationHelpers import handleIdsInMemory, handleFilterValuesInMemory
from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels
from .datamodelFeatureTrustee import TrusteePositionView
interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId) interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId)
if mode == "filterValues": if mode == "filterValues":
if not column: if not column:
@ -1507,7 +1504,6 @@ def delete_accounting_config(
"""Remove the accounting integration for this instance.""" """Remove the accounting integration for this instance."""
mandateId = _validateInstanceAccess(instanceId, context) mandateId = _validateInstanceAccess(instanceId, context)
interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId) interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId)
from .datamodelFeatureTrustee import TrusteeAccountingConfig
records = interface.db.getRecordset(TrusteeAccountingConfig, recordFilter={"featureInstanceId": instanceId}) records = interface.db.getRecordset(TrusteeAccountingConfig, recordFilter={"featureInstanceId": instanceId})
for r in records: for r in records:
interface.db.recordDelete(TrusteeAccountingConfig, r.get("id")) interface.db.recordDelete(TrusteeAccountingConfig, r.get("id"))
@ -1602,7 +1598,6 @@ def get_sync_status(
"""Get sync status of all positions for this instance.""" """Get sync status of all positions for this instance."""
mandateId = _validateInstanceAccess(instanceId, context) mandateId = _validateInstanceAccess(instanceId, context)
interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId) interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId)
from .datamodelFeatureTrustee import TrusteeAccountingSync
records = interface.db.getRecordset(TrusteeAccountingSync, recordFilter={"featureInstanceId": instanceId}) records = interface.db.getRecordset(TrusteeAccountingSync, recordFilter={"featureInstanceId": instanceId})
return {"items": [{k: v for k, v in r.items() if not k.startswith("_")} for r in records]} return {"items": [{k: v for k, v in r.items() if not k.startswith("_")} for r in records]}
@ -1618,7 +1613,6 @@ def get_position_sync_status(
"""Get sync status for a specific position.""" """Get sync status for a specific position."""
mandateId = _validateInstanceAccess(instanceId, context) mandateId = _validateInstanceAccess(instanceId, context)
interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId) interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId)
from .datamodelFeatureTrustee import TrusteeAccountingSync
records = interface.db.getRecordset(TrusteeAccountingSync, recordFilter={"positionId": positionId, "featureInstanceId": instanceId}) records = interface.db.getRecordset(TrusteeAccountingSync, recordFilter={"positionId": positionId, "featureInstanceId": instanceId})
return {"items": [{k: v for k, v in r.items() if not k.startswith("_")} for r in records]} return {"items": [{k: v for k, v in r.items() if not k.startswith("_")} for r in records]}
@ -1776,7 +1770,6 @@ def _serializeRoleForApi(role) -> Dict[str, Any]:
here (same pattern as ``getQuickActions``). Without this the React tree here (same pattern as ``getQuickActions``). Without this the React tree
crashes with "Objects are not valid as a React child". crashes with "Objects are not valid as a React child".
""" """
from modules.shared.i18nRegistry import resolveText
payload = role.model_dump() payload = role.model_dump()
payload["description"] = resolveText(payload.get("description")) payload["description"] = resolveText(payload.get("description"))
return payload return payload

1
modules/features/trustee/workflows/methodTrustee/actions/extractFromFiles.py
View file

@ -269,7 +269,6 @@ async def _extractWithAi(
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""3-step extraction: (1a) OCR/text via Vision AI, (1b) classify text, (2) structure by type.""" """3-step extraction: (1a) OCR/text via Vision AI, (1b) classify text, (2) structure by type."""
await self.services.ai.ensureAiObjectsInitialized() await self.services.ai.ensureAiObjectsInitialized()
from modules.datamodels.datamodelDocref import DocumentReferenceList, DocumentItemReference
docList = DocumentReferenceList( docList = DocumentReferenceList(
references=[DocumentItemReference(documentId=chatDocumentId, fileName=fileName)] references=[DocumentItemReference(documentId=chatDocumentId, fileName=fileName)]

1
modules/features/workspace/routeFeatureWorkspace.py
View file

@ -917,7 +917,6 @@ async def _runWorkspaceAgent(
messagePersisted = False messagePersisted = False
_toolSet = _cfg.get("toolSet", "core") _toolSet = _cfg.get("toolSet", "core")
_agentCfg = _cfg.get("agentConfig") _agentCfg = _cfg.get("agentConfig")
from modules.serviceCenter.services.serviceAgent.datamodelAgent import AgentConfig
agentCfgDict = dict(_agentCfg) if isinstance(_agentCfg, dict) else {} agentCfgDict = dict(_agentCfg) if isinstance(_agentCfg, dict) else {}
try: try:

2
modules/interfaces/interfaceAiObjects.py
View file

@ -465,7 +465,6 @@ class AiObjects:
toolChoice: Any = None, toolChoice: Any = None,
) -> AsyncGenerator[Union[str, AiCallResponse], None]: ) -> AsyncGenerator[Union[str, AiCallResponse], None]:
"""Stream a model call. Yields str deltas, then final AiCallResponse with billing.""" """Stream a model call. Yields str deltas, then final AiCallResponse with billing."""
from modules.datamodels.datamodelAi import AiModelCall, AiModelResponse
inputBytes = sum(len(str(m.get("content", "")).encode("utf-8")) for m in messages) inputBytes = sum(len(str(m.get("content", "")).encode("utf-8")) for m in messages)
startTime = time.time() startTime = time.time()
@ -537,7 +536,6 @@ class AiObjects:
Returns: Returns:
AiCallResponse with metadata["embeddings"] containing the vectors. AiCallResponse with metadata["embeddings"] containing the vectors.
""" """
from modules.aicore.aicoreBase import ContextLengthExceededException
if options is None: if options is None:
options = AiCallOptions(operationType=OperationTypeEnum.EMBEDDING) options = AiCallOptions(operationType=OperationTypeEnum.EMBEDDING)

18
modules/interfaces/interfaceDbApp.py
View file

@ -220,7 +220,6 @@ class AppObjects:
tableName = modelClass.__name__ tableName = modelClass.__name__
# Use buildDataObjectKey for semantic namespace lookup # Use buildDataObjectKey for semantic namespace lookup
from modules.interfaces.interfaceRbac import buildDataObjectKey
objectKey = buildDataObjectKey(tableName) objectKey = buildDataObjectKey(tableName)
permissions = self.rbac.getUserPermissions( permissions = self.rbac.getUserPermissions(
self.currentUser, self.currentUser,
@ -1122,8 +1121,6 @@ class AppObjects:
def _deleteUserReferencedData(self, userId: str) -> None: def _deleteUserReferencedData(self, userId: str) -> None:
"""Deletes all data associated with a user (full cascade).""" """Deletes all data associated with a user (full cascade)."""
try: try:
from modules.datamodels.datamodelNotification import UserNotification
from modules.datamodels.datamodelInvitation import Invitation
# 1. FeatureAccess + FeatureAccessRole # 1. FeatureAccess + FeatureAccessRole
accesses = self.db.getRecordset(FeatureAccess, recordFilter={"userId": userId}) accesses = self.db.getRecordset(FeatureAccess, recordFilter={"userId": userId})
@ -1560,7 +1557,6 @@ class AppObjects:
# Copy system template roles to new mandate (admin, user, viewer + AccessRules) # Copy system template roles to new mandate (admin, user, viewer + AccessRules)
try: try:
from modules.interfaces.interfaceRbac import copySystemRolesToMandate
copiedCount = copySystemRolesToMandate(self.db, mandateId) copiedCount = copySystemRolesToMandate(self.db, mandateId)
logger.info(f"Copied {copiedCount} system roles to new mandate {mandateId}") logger.info(f"Copied {copiedCount} system roles to new mandate {mandateId}")
except Exception as e: except Exception as e:
@ -1576,8 +1572,6 @@ class AppObjects:
``mandateLabel`` is the display name (Voller Name); a unique slug ``name`` (Kurzzeichen) is derived. ``mandateLabel`` is the display name (Voller Name); a unique slug ``name`` (Kurzzeichen) is derived.
""" """
from modules.datamodels.datamodelSubscription import MandateSubscription, SubscriptionStatusEnum, BUILTIN_PLANS from modules.datamodels.datamodelSubscription import MandateSubscription, SubscriptionStatusEnum, BUILTIN_PLANS
from modules.datamodels.datamodelFeatures import FeatureInstance
from modules.interfaces.interfaceRbac import copySystemRolesToMandate
from modules.interfaces.interfaceFeatures import getFeatureInterface from modules.interfaces.interfaceFeatures import getFeatureInterface
from modules.shared.featureDiscovery import loadFeatureMainModules from modules.shared.featureDiscovery import loadFeatureMainModules
plan = BUILTIN_PLANS.get(planKey) plan = BUILTIN_PLANS.get(planKey)
@ -1847,7 +1841,6 @@ class AppObjects:
raise PermissionError(f"No permission to delete mandate {mandateId}") raise PermissionError(f"No permission to delete mandate {mandateId}")
if not force: if not force:
from modules.shared.timeUtils import getUtcTimestamp
self.db.recordModify(Mandate, mandateId, {"enabled": False, "deletedAt": getUtcTimestamp()}) self.db.recordModify(Mandate, mandateId, {"enabled": False, "deletedAt": getUtcTimestamp()})
logger.info(f"Soft-deleted mandate {mandateId} (30-day retention)") logger.info(f"Soft-deleted mandate {mandateId} (30-day retention)")
return True return True
@ -1858,8 +1851,6 @@ class AppObjects:
from modules.datamodels.datamodelFiles import FileItem from modules.datamodels.datamodelFiles import FileItem
from modules.datamodels.datamodelDataSource import DataSource from modules.datamodels.datamodelDataSource import DataSource
from modules.datamodels.datamodelKnowledge import FileContentIndex, ContentChunk from modules.datamodels.datamodelKnowledge import FileContentIndex, ContentChunk
from modules.datamodels.datamodelFeatures import FeatureDataSource
from modules.datamodels.datamodelFeatures import DataNeutralizerAttributes
instances = self.db.getRecordset(FeatureInstance, recordFilter={"mandateId": mandateId}) instances = self.db.getRecordset(FeatureInstance, recordFilter={"mandateId": mandateId})
@ -1983,7 +1974,6 @@ class AppObjects:
# 3b. Billing data cascade handled by onMandateDelete lifecycle hook (interfaceDbBilling) # 3b. Billing data cascade handled by onMandateDelete lifecycle hook (interfaceDbBilling)
# 3c. Delete Invitations for this mandate # 3c. Delete Invitations for this mandate
from modules.datamodels.datamodelInvitation import Invitation
invitations = self.db.getRecordset(Invitation, recordFilter={"mandateId": mandateId}) invitations = self.db.getRecordset(Invitation, recordFilter={"mandateId": mandateId})
for inv in invitations: for inv in invitations:
self.db.recordDelete(Invitation, inv.get("id")) self.db.recordDelete(Invitation, inv.get("id"))
@ -1991,7 +1981,6 @@ class AppObjects:
logger.info(f"Cascade: deleted {len(invitations)} Invitations for mandate {mandateId}") logger.info(f"Cascade: deleted {len(invitations)} Invitations for mandate {mandateId}")
# 4. Delete mandate-level Roles # 4. Delete mandate-level Roles
from modules.datamodels.datamodelRbac import Role, AccessRule
roles = self.db.getRecordset(Role, recordFilter={"mandateId": mandateId}) roles = self.db.getRecordset(Role, recordFilter={"mandateId": mandateId})
for role in roles: for role in roles:
rules = self.db.getRecordset(AccessRule, recordFilter={"roleId": role.get("id")}) rules = self.db.getRecordset(AccessRule, recordFilter={"roleId": role.get("id")})
@ -3961,7 +3950,6 @@ class AppObjects:
def getTableListViews(self, contextKey: str) -> list: def getTableListViews(self, contextKey: str) -> list:
"""Return all saved views for the current user and contextKey.""" """Return all saved views for the current user and contextKey."""
from modules.datamodels.datamodelPagination import TableListView
try: try:
rows = self.db.getRecordset( rows = self.db.getRecordset(
TableListView, TableListView,
@ -3980,7 +3968,6 @@ class AppObjects:
def getTableListView(self, contextKey: str, viewKey: str): def getTableListView(self, contextKey: str, viewKey: str):
"""Return one view by viewKey or None if not found.""" """Return one view by viewKey or None if not found."""
from modules.datamodels.datamodelPagination import TableListView
try: try:
rows = self.db.getRecordset( rows = self.db.getRecordset(
TableListView, TableListView,
@ -3996,8 +3983,6 @@ class AppObjects:
def createTableListView(self, contextKey: str, viewKey: str, displayName: str, config: dict): def createTableListView(self, contextKey: str, viewKey: str, displayName: str, config: dict):
"""Create a new view. Raises ValueError if viewKey already exists for this context.""" """Create a new view. Raises ValueError if viewKey already exists for this context."""
from modules.datamodels.datamodelPagination import TableListView
from modules.shared.timeUtils import getUtcTimestamp
if self.getTableListView(contextKey=contextKey, viewKey=viewKey) is not None: if self.getTableListView(contextKey=contextKey, viewKey=viewKey) is not None:
raise ValueError(f"View '{viewKey}' already exists for context '{contextKey}'") raise ValueError(f"View '{viewKey}' already exists for context '{contextKey}'")
data = { data = {
@ -4018,8 +4003,6 @@ class AppObjects:
def updateTableListView(self, viewId: str, updates: dict): def updateTableListView(self, viewId: str, updates: dict):
"""Update an existing view by its primary key id.""" """Update an existing view by its primary key id."""
from modules.datamodels.datamodelPagination import TableListView
from modules.shared.timeUtils import getUtcTimestamp
try: try:
updates = {**updates, "updatedAt": getUtcTimestamp()} updates = {**updates, "updatedAt": getUtcTimestamp()}
self.db.recordModify(TableListView, viewId, updates) self.db.recordModify(TableListView, viewId, updates)
@ -4034,7 +4017,6 @@ class AppObjects:
def deleteTableListView(self, viewId: str) -> bool: def deleteTableListView(self, viewId: str) -> bool:
"""Delete a view by primary key id. Returns True on success.""" """Delete a view by primary key id. Returns True on success."""
from modules.datamodels.datamodelPagination import TableListView
try: try:
self.db.recordDelete(TableListView, viewId) self.db.recordDelete(TableListView, viewId)
return True return True

2
modules/interfaces/interfaceDbBilling.py
View file

@ -1654,8 +1654,6 @@ class BillingObjects:
`amount` column. Resolves matching mandate/user IDs via the app DB `amount` column. Resolves matching mandate/user IDs via the app DB
first, then builds a single SQL query with OR-combined conditions. first, then builds a single SQL query with OR-combined conditions.
""" """
from modules.connectors.connectorDbPostgre import getModelFields, parseRecordFields
from modules.datamodels.datamodelUam import UserInDB
from modules.interfaces.interfaceDbApp import getInterface as getAppInterface from modules.interfaces.interfaceDbApp import getInterface as getAppInterface
table = BillingTransaction.__name__ table = BillingTransaction.__name__

2
modules/interfaces/interfaceDbChat.py
View file

@ -393,7 +393,6 @@ class ChatObjects:
tableName = modelClass.__name__ tableName = modelClass.__name__
# Use buildDataObjectKey for semantic namespace lookup # Use buildDataObjectKey for semantic namespace lookup
from modules.interfaces.interfaceRbac import buildDataObjectKey
objectKey = buildDataObjectKey(tableName) objectKey = buildDataObjectKey(tableName)
permissions = self.rbac.getUserPermissions( permissions = self.rbac.getUserPermissions(
self.currentUser, self.currentUser,
@ -826,7 +825,6 @@ class ChatObjects:
if not effectiveMandateId: if not effectiveMandateId:
# Fall back to Root mandate (first mandate in system) # Fall back to Root mandate (first mandate in system)
try: try:
from modules.datamodels.datamodelUam import Mandate
from modules.security.rootAccess import getRootDbAppConnector from modules.security.rootAccess import getRootDbAppConnector
dbAppConn = getRootDbAppConnector() dbAppConn = getRootDbAppConnector()
allMandates = dbAppConn.getRecordset(Mandate) allMandates = dbAppConn.getRecordset(Mandate)

1
modules/interfaces/interfaceDbKnowledge.py
View file

@ -741,7 +741,6 @@ def migrateVectorDimensions():
If it differs from the target, nulls existing embeddings and alters the column type. If it differs from the target, nulls existing embeddings and alters the column type.
Safe to call on every startup — skips when dimensions already match or table doesn't exist. Safe to call on every startup — skips when dimensions already match or table doesn't exist.
""" """
from modules.datamodels.datamodelKnowledge import KNOWLEDGE_EMBEDDING_DIMENSIONS
targetDim = KNOWLEDGE_EMBEDDING_DIMENSIONS targetDim = KNOWLEDGE_EMBEDDING_DIMENSIONS
interface = getInterface() interface = getInterface()

5
modules/interfaces/interfaceDbManagement.py
View file

@ -317,7 +317,6 @@ class ComponentObjects:
return False return False
tableName = modelClass.__name__ tableName = modelClass.__name__
from modules.interfaces.interfaceRbac import buildDataObjectKey
objectKey = buildDataObjectKey(tableName) objectKey = buildDataObjectKey(tableName)
permissions = self.rbac.getUserPermissions( permissions = self.rbac.getUserPermissions(
self.currentUser, self.currentUser,
@ -1066,7 +1065,6 @@ class ComponentObjects:
Owners always can. Non-owners need RBAC ALL level.""" Owners always can. Non-owners need RBAC ALL level."""
if self._isFolderOwner(folder): if self._isFolderOwner(folder):
return return
from modules.interfaces.interfaceRbac import buildDataObjectKey
objectKey = buildDataObjectKey("FileFolder") objectKey = buildDataObjectKey("FileFolder")
permissions = self.rbac.getUserPermissions( permissions = self.rbac.getUserPermissions(
self.currentUser, AccessRuleContext.DATA, objectKey, self.currentUser, AccessRuleContext.DATA, objectKey,
@ -1207,7 +1205,6 @@ class ComponentObjects:
self._requireFolderWriteAccess(folder, folderId, "update") self._requireFolderWriteAccess(folder, folderId, "update")
if scope == "global": if scope == "global":
from modules.interfaces.interfaceRbac import buildDataObjectKey
objectKey = buildDataObjectKey("FileFolder") objectKey = buildDataObjectKey("FileFolder")
permissions = self.rbac.getUserPermissions( permissions = self.rbac.getUserPermissions(
self.currentUser, AccessRuleContext.DATA, objectKey, self.currentUser, AccessRuleContext.DATA, objectKey,
@ -1387,8 +1384,6 @@ class ComponentObjects:
Owners always can. Non-owners need RBAC ALL level.""" Owners always can. Non-owners need RBAC ALL level."""
if self._isFileOwner(file): if self._isFileOwner(file):
return return
from modules.interfaces.interfaceRbac import buildDataObjectKey
from modules.datamodels.datamodelRbac import AccessRuleContext
objectKey = buildDataObjectKey("FileItem") objectKey = buildDataObjectKey("FileItem")
permissions = self.rbac.getUserPermissions( permissions = self.rbac.getUserPermissions(
self.currentUser, AccessRuleContext.DATA, objectKey, self.currentUser, AccessRuleContext.DATA, objectKey,

6
modules/interfaces/interfaceRbac.py
View file

@ -379,7 +379,6 @@ def getRecordsetWithRBAC(
# Handle JSONB fields and ensure numeric types are correct # Handle JSONB fields and ensure numeric types are correct
# Import the helper function from connector module # Import the helper function from connector module
from modules.connectors.connectorDbPostgre import getModelFields
fields = getModelFields(modelClass) fields = getModelFields(modelClass)
for record in records: for record in records:
for fieldName, fieldType in fields.items(): for fieldName, fieldType in fields.items():
@ -511,7 +510,6 @@ def getRecordsetPaginatedWithRBAC(
whereValues.append(value) whereValues.append(value)
if pagination and pagination.filters: if pagination and pagination.filters:
from modules.connectors.connectorDbPostgre import getModelFields
fields = getModelFields(modelClass) fields = getModelFields(modelClass)
validColumns = set(fields.keys()) validColumns = set(fields.keys())
for key, val in pagination.filters.items(): for key, val in pagination.filters.items():
@ -545,7 +543,6 @@ def getRecordsetPaginatedWithRBAC(
orderParts: List[str] = [] orderParts: List[str] = []
if pagination and pagination.sort: if pagination and pagination.sort:
from modules.connectors.connectorDbPostgre import getModelFields
validColumns = set(getModelFields(modelClass).keys()) validColumns = set(getModelFields(modelClass).keys())
for sf in pagination.sort: for sf in pagination.sort:
if sf.field in validColumns: if sf.field in validColumns:
@ -569,7 +566,6 @@ def getRecordsetPaginatedWithRBAC(
cursor.execute(dataSql, whereValues) cursor.execute(dataSql, whereValues)
records = [dict(row) for row in cursor.fetchall()] records = [dict(row) for row in cursor.fetchall()]
from modules.connectors.connectorDbPostgre import getModelFields, parseRecordFields
fields = getModelFields(modelClass) fields = getModelFields(modelClass)
for record in records: for record in records:
parseRecordFields(record, fields, f"table {table}") parseRecordFields(record, fields, f"table {table}")
@ -625,7 +621,6 @@ def getDistinctColumnValuesWithRBAC(
if not connector._ensureTableExists(modelClass): if not connector._ensureTableExists(modelClass):
return [] return []
from modules.connectors.connectorDbPostgre import getModelFields
fields = getModelFields(modelClass) fields = getModelFields(modelClass)
if column not in fields: if column not in fields:
return [] return []
@ -949,7 +944,6 @@ def buildRbacWhereClause(
# Fall back to Root mandate (first mandate in system) for GROUP access # Fall back to Root mandate (first mandate in system) for GROUP access
# This allows system-level tables to be accessed without explicit mandate context # This allows system-level tables to be accessed without explicit mandate context
try: try:
from modules.datamodels.datamodelUam import Mandate
dbApp = getRootDbAppConnector() dbApp = getRootDbAppConnector()
allMandates = dbApp.getRecordset(Mandate) allMandates = dbApp.getRecordset(Mandate)
if allMandates: if allMandates:

2
modules/interfaces/interfaceTableHelpers.py
View file

@ -85,7 +85,6 @@ def applyViewToParams(params: Optional["PaginationParams"], viewConfig: Optional
Returns the (mutated) params, or a new minimal PaginationParams when Returns the (mutated) params, or a new minimal PaginationParams when
params is None (so callers always get a valid object). params is None (so callers always get a valid object).
""" """
from modules.datamodels.datamodelPagination import SortField
if not viewConfig: if not viewConfig:
return params return params
@ -264,7 +263,6 @@ def buildGroupLayout(
------- -------
(page_items, GroupLayout | None) (page_items, GroupLayout | None)
""" """
from modules.datamodels.datamodelPagination import GroupBand, GroupLayout
if not groupByLevels: if not groupByLevels:
offset = (page - 1) * pageSize offset = (page - 1) * pageSize

1
modules/routes/routeAdminFeatures.py
View file

@ -473,7 +473,6 @@ def list_feature_instances(
items = [inst.model_dump() for inst in instances] items = [inst.model_dump() for inst in instances]
from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels
from modules.datamodels.datamodelFeatures import FeatureInstance
enrichRowsWithFkLabels(items, FeatureInstance, db=rootInterface.db) enrichRowsWithFkLabels(items, FeatureInstance, db=rootInterface.db)
if mode == "filterValues": if mode == "filterValues":

3
modules/routes/routeAdminRbacRules.py
View file

@ -40,7 +40,6 @@ def _getAdminMandateIds(context: RequestContext) -> List[str]:
"""Get mandate IDs where the user has an admin role.""" """Get mandate IDs where the user has an admin role."""
mandateIds = [] mandateIds = []
try: try:
from modules.interfaces.interfaceDbApp import getRootInterface
rootInterface = getRootInterface() rootInterface = getRootInterface()
userMandates = rootInterface.getUserMandates(str(context.user.id)) userMandates = rootInterface.getUserMandates(str(context.user.id))
for um in userMandates: for um in userMandates:
@ -64,7 +63,6 @@ def _getAdminMandateIds(context: RequestContext) -> List[str]:
def _isRoleInAdminMandates(roleId: str, adminMandateIds: List[str]) -> bool: def _isRoleInAdminMandates(roleId: str, adminMandateIds: List[str]) -> bool:
"""Check if a role belongs to one of the admin's mandates.""" """Check if a role belongs to one of the admin's mandates."""
try: try:
from modules.interfaces.interfaceDbApp import getRootInterface
rootInterface = getRootInterface() rootInterface = getRootInterface()
role = rootInterface.getRole(roleId) role = rootInterface.getRole(roleId)
if not role: if not role:
@ -1405,7 +1403,6 @@ def cleanup_duplicate_access_rules(
# Phase 2: Fix template role assignments # Phase 2: Fix template role assignments
# UserMandateRole should reference mandate-instance roles, not templates # UserMandateRole should reference mandate-instance roles, not templates
# ===================================================================== # =====================================================================
from modules.datamodels.datamodelMembership import UserMandate, UserMandateRole
allUserMandateRoles = rootInterface.db.getRecordset(UserMandateRole) allUserMandateRoles = rootInterface.db.getRecordset(UserMandateRole)
templateFixDetails = [] templateFixDetails = []

2
modules/routes/routeBilling.py
View file

@ -756,7 +756,6 @@ def createOrUpdateSettings(
return result or existingSettings return result or existingSettings
return existingSettings return existingSettings
else: else:
from modules.datamodels.datamodelBilling import BillingSettings
newSettings = BillingSettings( newSettings = BillingSettings(
mandateId=targetMandateId, mandateId=targetMandateId,
@ -821,7 +820,6 @@ def addCredit(
if creditRequest.amount == 0: if creditRequest.amount == 0:
raise HTTPException(status_code=400, detail=routeApiMsg("Amount must not be zero")) raise HTTPException(status_code=400, detail=routeApiMsg("Amount must not be zero"))
from modules.datamodels.datamodelBilling import BillingTransaction
isDeduction = creditRequest.amount < 0 isDeduction = creditRequest.amount < 0
txType = TransactionTypeEnum.DEBIT if isDeduction else TransactionTypeEnum.CREDIT txType = TransactionTypeEnum.DEBIT if isDeduction else TransactionTypeEnum.CREDIT

3
modules/routes/routeDataConnections.py
View file

@ -161,7 +161,6 @@ async def get_connections(
from modules.interfaces.interfaceTableHelpers import ( from modules.interfaces.interfaceTableHelpers import (
resolveView, applyViewToParams, buildGroupLayout, effective_group_by_levels, resolveView, applyViewToParams, buildGroupLayout, effective_group_by_levels,
) )
from modules.datamodels.datamodelPagination import AppliedViewMeta
CONTEXT_KEY = "connections" CONTEXT_KEY = "connections"
@ -782,7 +781,6 @@ async def _updateKnowledgeConsent(
if not connection: if not connection:
raise HTTPException(status_code=404, detail=routeApiMsg("Connection not found")) raise HTTPException(status_code=404, detail=routeApiMsg("Connection not found"))
from modules.interfaces.interfaceDbApp import getRootInterface
rootIf = getRootInterface() rootIf = getRootInterface()
rootIf.db.recordModify(UserConnection, connectionId, {"knowledgeIngestionEnabled": enabled}) rootIf.db.recordModify(UserConnection, connectionId, {"knowledgeIngestionEnabled": enabled})
@ -861,7 +859,6 @@ def _updateKnowledgePreferences(
cleaned = {k: v for k, v in preferences.items() if k in _ALLOWED_KEYS} cleaned = {k: v for k, v in preferences.items() if k in _ALLOWED_KEYS}
merged = {**existing, **cleaned, "schemaVersion": 1} merged = {**existing, **cleaned, "schemaVersion": 1}
from modules.interfaces.interfaceDbApp import getRootInterface
getRootInterface().db.recordModify(UserConnection, connectionId, {"knowledgePreferences": merged}) getRootInterface().db.recordModify(UserConnection, connectionId, {"knowledgePreferences": merged})
logger.info("Knowledge preferences updated for connection %s", connectionId) logger.info("Knowledge preferences updated for connection %s", connectionId)

2
modules/routes/routeDataFiles.py
View file

@ -738,7 +738,6 @@ def get_files(
resolveView, applyViewToParams, buildGroupLayout, effective_group_by_levels, resolveView, applyViewToParams, buildGroupLayout, effective_group_by_levels,
) )
import modules.interfaces.interfaceDbApp as _appIface import modules.interfaces.interfaceDbApp as _appIface
from modules.datamodels.datamodelPagination import AppliedViewMeta
managementInterface = interfaceDbManagement.getInterface( managementInterface = interfaceDbManagement.getInterface(
currentUser, currentUser,
@ -1202,7 +1201,6 @@ def bulk_set_neutralize(
managementInterface.updateFile(fid, {"neutralize": neutralize}) managementInterface.updateFile(fid, {"neutralize": neutralize})
if not neutralize: if not neutralize:
try: try:
from modules.interfaces import interfaceDbKnowledge
kIface = interfaceDbKnowledge.getInterface(currentUser) kIface = interfaceDbKnowledge.getInterface(currentUser)
kIface.purgeFileKnowledge(fid) kIface.purgeFileKnowledge(fid)
except Exception as ke: except Exception as ke:

1
modules/routes/routeDataPrompts.py
View file

@ -55,7 +55,6 @@ def get_prompts(
resolveView, applyViewToParams, buildGroupLayout, effective_group_by_levels, resolveView, applyViewToParams, buildGroupLayout, effective_group_by_levels,
) )
from modules.interfaces.interfaceDbApp import getInterface as getAppInterface from modules.interfaces.interfaceDbApp import getInterface as getAppInterface
from modules.datamodels.datamodelPagination import AppliedViewMeta
CONTEXT_KEY = "prompts" CONTEXT_KEY = "prompts"

1
modules/routes/routeMfa.py
View file

@ -215,7 +215,6 @@ def mfaVerify(
jti = jwt.decode(accessToken, SECRET_KEY, algorithms=[ALGORITHM]).get("jti") jti = jwt.decode(accessToken, SECRET_KEY, algorithms=[ALGORITHM]).get("jti")
from modules.interfaces.interfaceDbApp import getInterface
user = User.model_validate(userRecord) user = User.model_validate(userRecord)
userInterface = getInterface(user) userInterface = getInterface(user)
dbToken = Token( dbToken = Token(

1
modules/routes/routeNotifications.py
View file

@ -411,7 +411,6 @@ def _handleInvitationAction(
) -> str: ) -> str:
"""Handle accept/decline actions for invitation notifications.""" """Handle accept/decline actions for invitation notifications."""
from modules.datamodels.datamodelInvitation import Invitation from modules.datamodels.datamodelInvitation import Invitation
from modules.datamodels.datamodelUam import Mandate
from modules.datamodels.datamodelMembership import UserMandate from modules.datamodels.datamodelMembership import UserMandate
invitationId = notification.referenceId invitationId = notification.referenceId

1
modules/routes/routeRagInventory.py
View file

@ -485,7 +485,6 @@ def _getInventoryPlatform(
from modules.interfaces.interfaceDbApp import getRootInterface from modules.interfaces.interfaceDbApp import getRootInterface
from modules.interfaces.interfaceDbKnowledge import getInterface as getKnowledgeInterface from modules.interfaces.interfaceDbKnowledge import getInterface as getKnowledgeInterface
from modules.serviceCenter.services.serviceBackgroundJobs import mainBackgroundJobService as jobService from modules.serviceCenter.services.serviceBackgroundJobs import mainBackgroundJobService as jobService
from modules.datamodels.datamodelUam import UserConnection
rootIf = getRootInterface() rootIf = getRootInterface()
knowledgeIf = getKnowledgeInterface(None) knowledgeIf = getKnowledgeInterface(None)

4
modules/routes/routeSecurityLocal.py
View file

@ -70,7 +70,6 @@ def buildAuthEmailHtml(
operatorLine = "" operatorLine = ""
try: try:
from modules.shared.configuration import APP_CONFIG
parts = [p for p in [ parts = [p for p in [
APP_CONFIG.get("Operator_CompanyName", ""), APP_CONFIG.get("Operator_CompanyName", ""),
APP_CONFIG.get("Operator_Address", ""), APP_CONFIG.get("Operator_Address", ""),
@ -194,7 +193,6 @@ def _ensureHomeMandate(rootInterface, user) -> None:
return return
try: try:
from modules.interfaces.interfaceDbApp import getRootInterface as _getRootIf
appIf = _getRootIf() appIf = _getRootIf()
normalizedEmail = (user.email or "").strip().lower() if user.email else None normalizedEmail = (user.email or "").strip().lower() if user.email else None
pendingByUsername = appIf.getInvitationsByTargetUsername(user.username) pendingByUsername = appIf.getInvitationsByTargetUsername(user.username)
@ -1058,7 +1056,6 @@ def _getNeutralizationMappings(
): ):
"""List the current user's neutralization placeholder mappings.""" """List the current user's neutralization placeholder mappings."""
userId = str(context.user.id) userId = str(context.user.id)
from modules.interfaces.interfaceDbApp import getRootInterface
from modules.datamodels.datamodelFeatures import DataNeutralizerAttributes from modules.datamodels.datamodelFeatures import DataNeutralizerAttributes
rootIf = getRootInterface() rootIf = getRootInterface()
records = rootIf.db.getRecordset(DataNeutralizerAttributes, recordFilter={"userId": userId}) records = rootIf.db.getRecordset(DataNeutralizerAttributes, recordFilter={"userId": userId})
@ -1074,7 +1071,6 @@ def _deleteNeutralizationMapping(
): ):
"""Delete a specific neutralization mapping owned by the current user.""" """Delete a specific neutralization mapping owned by the current user."""
userId = str(context.user.id) userId = str(context.user.id)
from modules.interfaces.interfaceDbApp import getRootInterface
from modules.datamodels.datamodelFeatures import DataNeutralizerAttributes from modules.datamodels.datamodelFeatures import DataNeutralizerAttributes
rootIf = getRootInterface() rootIf = getRootInterface()
records = rootIf.db.getRecordset(DataNeutralizerAttributes, recordFilter={"id": mappingId}) records = rootIf.db.getRecordset(DataNeutralizerAttributes, recordFilter={"id": mappingId})

7
modules/routes/routeWorkflowAutomation.py
View file

@ -64,7 +64,6 @@ async def _listWorkflows(
mandateId: Optional[str] = Query(default=None), mandateId: Optional[str] = Query(default=None),
): ):
from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels
from modules.interfaces.interfaceDbApp import getRootInterface as _getRootIface
db = _getWorkflowAutomationDb() db = _getWorkflowAutomationDb()
try: try:
db._ensureTableExists(AutoWorkflow) db._ensureTableExists(AutoWorkflow)
@ -174,7 +173,6 @@ async def _listRuns(
workflowId: Optional[str] = Query(default=None), workflowId: Optional[str] = Query(default=None),
): ):
from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels
from modules.interfaces.interfaceDbApp import getRootInterface as _getRootIface
db = _getWorkflowAutomationDb() db = _getWorkflowAutomationDb()
try: try:
db._ensureTableExists(AutoRun) db._ensureTableExists(AutoRun)
@ -476,17 +474,14 @@ def _listTemplates(
templates = iface.getTemplates(scope=scope) templates = iface.getTemplates(scope=scope)
from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels
from modules.interfaces.interfaceDbApp import getRootInterface as _getRootIface
enrichRowsWithFkLabels(templates, AutoWorkflow, db=_getRootIface().db) enrichRowsWithFkLabels(templates, AutoWorkflow, db=_getRootIface().db)
if mode == "filterValues": if mode == "filterValues":
if not column: if not column:
raise HTTPException(status_code=400, detail="column parameter required for mode=filterValues") raise HTTPException(status_code=400, detail="column parameter required for mode=filterValues")
from modules.dbHelpers.paginationHelpers import handleFilterValuesInMemory
return handleFilterValuesInMemory(templates, column, pagination) return handleFilterValuesInMemory(templates, column, pagination)
if mode == "ids": if mode == "ids":
from modules.dbHelpers.paginationHelpers import handleIdsInMemory
return handleIdsInMemory(templates, pagination) return handleIdsInMemory(templates, pagination)
paginationParams = None paginationParams = None
@ -1328,7 +1323,6 @@ def _getRunDetail(
if tid: if tid:
try: try:
from modules.dbHelpers.fkLabelResolver import resolveInstanceLabels from modules.dbHelpers.fkLabelResolver import resolveInstanceLabels
from modules.interfaces.interfaceDbApp import getRootInterface as _getRootIface
labelMap = resolveInstanceLabels(_getRootIface().db, [tid]) labelMap = resolveInstanceLabels(_getRootIface().db, [tid])
targetInstanceLabel = labelMap.get(tid) targetInstanceLabel = labelMap.get(tid)
except Exception: except Exception:
@ -1425,7 +1419,6 @@ def _startEmailPollerIfNeeded(result: dict) -> None:
if not isinstance(result, dict) or result.get("waitReason") != "email": if not isinstance(result, dict) or result.get("waitReason") != "email":
return return
try: try:
from modules.interfaces.interfaceDbApp import getRootInterface
from modules.workflowAutomation.scheduler.emailPoller import ensureRunning from modules.workflowAutomation.scheduler.emailPoller import ensureRunning
root = getRootInterface() root = getRootInterface()
eventUser = root.getUserByUsername("event") if root else None eventUser = root.getUserByUsername("event") if root else None

7
modules/serviceCenter/services/serviceAgent/mainServiceAgent.py
View file

@ -425,7 +425,6 @@ class AgentService:
activeToolNames.update(tb.tools) activeToolNames.update(tb.tools)
from modules.serviceCenter.services.serviceAgent.externalToolRegistry import getExternalTools from modules.serviceCenter.services.serviceAgent.externalToolRegistry import getExternalTools
from modules.serviceCenter.services.serviceAgent.datamodelAgent import ToolDefinition
for tb in activeToolboxes: for tb in activeToolboxes:
extDefs = getExternalTools(tb.id) extDefs = getExternalTools(tb.id)
if not extDefs: if not extDefs:
@ -459,7 +458,6 @@ class AgentService:
from modules.serviceCenter.services.serviceAgent.toolboxRegistry import ( from modules.serviceCenter.services.serviceAgent.toolboxRegistry import (
getToolboxRegistry, buildRequestToolboxDefinition, REQUEST_TOOLBOX_TOOL_NAME, getToolboxRegistry, buildRequestToolboxDefinition, REQUEST_TOOLBOX_TOOL_NAME,
) )
from modules.serviceCenter.services.serviceAgent.datamodelAgent import ToolResult
tbRegistry = getToolboxRegistry() tbRegistry = getToolboxRegistry()
allIds = [tb.id for tb in tbRegistry.getAllToolboxes()] allIds = [tb.id for tb in tbRegistry.getAllToolboxes()]
@ -488,7 +486,6 @@ class AgentService:
activatedCount += 1 activatedCount += 1
continue continue
try: try:
from modules.serviceCenter.services.serviceAgent.coreTools import registerCoreTools
registerCoreTools(registry, self.services) registerCoreTools(registry, self.services)
if registry.isValidTool(toolName): if registry.isValidTool(toolName):
activatedCount += 1 activatedCount += 1
@ -499,9 +496,6 @@ class AgentService:
try: try:
from modules.workflows.processing.shared.methodDiscovery import discoverMethods from modules.workflows.processing.shared.methodDiscovery import discoverMethods
from modules.workflows.processing.core.actionExecutor import ActionExecutor from modules.workflows.processing.core.actionExecutor import ActionExecutor
from modules.serviceCenter.services.serviceAgent.actionToolAdapter import (
ActionToolAdapter,
)
discoverMethods(self.services) discoverMethods(self.services)
adapter = ActionToolAdapter(ActionExecutor(self.services)) adapter = ActionToolAdapter(ActionExecutor(self.services))
@ -622,7 +616,6 @@ class AgentService:
def _createPersistRoundMemoryFn(self, workflowId: str): def _createPersistRoundMemoryFn(self, workflowId: str):
"""Create callback that persists RoundMemory entries after tool execution.""" """Create callback that persists RoundMemory entries after tool execution."""
from modules.serviceCenter.services.serviceAgent.agentLoop import classifyToolResult
from modules.datamodels.datamodelKnowledge import RoundMemory from modules.datamodels.datamodelKnowledge import RoundMemory
async def _persistRoundMemory( async def _persistRoundMemory(

2
modules/serviceCenter/services/serviceAi/mainServiceAi.py
View file

@ -335,7 +335,6 @@ class AiService:
Returns: Returns:
AiCallResponse with content as JSON string (SpeechTeamsResponse format) AiCallResponse with content as JSON string (SpeechTeamsResponse format)
""" """
from modules.datamodels.datamodelAi import AiCallResponse, AiModelCall, AiCallOptions, PriorityEnum
startTime = time.time() startTime = time.time()
@ -637,7 +636,6 @@ detectedIntent-Werte:
try: try:
from modules.aicore.aicoreModelRegistry import modelRegistry from modules.aicore.aicoreModelRegistry import modelRegistry
from modules.aicore.aicoreModelSelector import modelSelector as _modSel from modules.aicore.aicoreModelSelector import modelSelector as _modSel
from modules.datamodels.datamodelAi import AiCallOptions, OperationTypeEnum
_models = modelRegistry.getAvailableModels() _models = modelRegistry.getAvailableModels()
_providers = self._calculateEffectiveProviders() _providers = self._calculateEffectiveProviders()

2
modules/serviceCenter/services/serviceChat/mainServiceChat.py
View file

@ -615,7 +615,6 @@ class ChatService:
def getUserVoicePreferences(self, userId: str, mandateId: str = None) -> Optional[Dict[str, Any]]: def getUserVoicePreferences(self, userId: str, mandateId: str = None) -> Optional[Dict[str, Any]]:
"""Get TTS voice preferences for a user, resolved by mandate scope.""" """Get TTS voice preferences for a user, resolved by mandate scope."""
from modules.datamodels.datamodelUam import UserVoicePreferences
try: try:
prefRecords = self.interfaceDbApp.db.getRecordset( prefRecords = self.interfaceDbApp.db.getRecordset(
UserVoicePreferences, recordFilter={"userId": userId} UserVoicePreferences, recordFilter={"userId": userId}
@ -842,7 +841,6 @@ class ChatService:
"""Create an ActionItem record in the chat DB. """Create an ActionItem record in the chat DB.
Encapsulates low-level _separateObjectFields + db.recordCreate so callers Encapsulates low-level _separateObjectFields + db.recordCreate so callers
never need direct interfaceDbChat access.""" never need direct interfaceDbChat access."""
from modules.datamodels.datamodelChat import ActionItem
simpleFields, _objectFields = self.interfaceDbChat._separateObjectFields(ActionItem, actionData) simpleFields, _objectFields = self.interfaceDbChat._separateObjectFields(ActionItem, actionData)
return self.interfaceDbChat.db.recordCreate(ActionItem, simpleFields) return self.interfaceDbChat.db.recordCreate(ActionItem, simpleFields)

1
modules/serviceCenter/services/serviceExtraction/extractors/extractorContainer.py
View file

@ -204,7 +204,6 @@ def _addFilePart(
entryPath = f"{containerPath}/{fileName}" if containerPath else fileName entryPath = f"{containerPath}/{fileName}" if containerPath else fileName
detectedMime = _detectMimeType(fileName) detectedMime = _detectMimeType(fileName)
from ..subRegistry import getExtractorRegistry
registry = getExtractorRegistry() registry = getExtractorRegistry()
extractor = registry.resolve(detectedMime, fileName) extractor = registry.resolve(detectedMime, fileName)

1
modules/serviceCenter/services/serviceExtraction/extractors/extractorEmail.py
View file

@ -255,7 +255,6 @@ def _delegateAttachment(attachData: bytes, attachName: str, parentId: str, depth
guessedMime, _ = mimetypes.guess_type(attachName) guessedMime, _ = mimetypes.guess_type(attachName)
detectedMime = guessedMime or "application/octet-stream" detectedMime = guessedMime or "application/octet-stream"
from ..subRegistry import getExtractorRegistry
registry = getExtractorRegistry() registry = getExtractorRegistry()
extractor = registry.resolve(detectedMime, attachName) extractor = registry.resolve(detectedMime, attachName)

1
modules/serviceCenter/services/serviceExtraction/extractors/extractorFolder.py
View file

@ -141,7 +141,6 @@ def _walkFolder(
guessedMime, _ = mimetypes.guess_type(entry.name) guessedMime, _ = mimetypes.guess_type(entry.name)
detectedMime = guessedMime or "application/octet-stream" detectedMime = guessedMime or "application/octet-stream"
from ..subRegistry import ExtractorRegistry
registry = ExtractorRegistry() registry = ExtractorRegistry()
extractor = registry.resolve(detectedMime, entry.name) extractor = registry.resolve(detectedMime, entry.name)

2
modules/serviceCenter/services/serviceExtraction/subRegistry.py
View file

@ -50,8 +50,6 @@ class Extractor:
precomputedParts: Optional[List[ContentPart]] = None, precomputedParts: Optional[List[ContentPart]] = None,
) -> "UdmDocument": ) -> "UdmDocument":
"""Build UDM from extracted parts (default: heuristic grouping). Override for format-specific trees.""" """Build UDM from extracted parts (default: heuristic grouping). Override for format-specific trees."""
from modules.datamodels.datamodelUdm import contentPartsToUdm, mimeToUdmSourceType
from modules.datamodels.datamodelExtraction import ContentExtracted
from .subUtils import makeId from .subUtils import makeId
parts = precomputedParts if precomputedParts is not None else self.extract(fileBytes, context) parts = precomputedParts if precomputedParts is not None else self.extract(fileBytes, context)

1
modules/serviceCenter/services/serviceGeneration/renderers/rendererPdf.py
View file

@ -230,7 +230,6 @@ class RendererPdf(BaseRenderer):
# memory simultaneously. Collected here, deleted after the build. # memory simultaneously. Collected here, deleted after the build.
self._tempImageFiles = [] self._tempImageFiles = []
try: try:
from modules.serviceCenter.services.serviceGeneration.styleDefaults import resolveStyle
self._unifiedStyle = unifiedStyle or resolveStyle(None) self._unifiedStyle = unifiedStyle or resolveStyle(None)
styles = self._convertUnifiedStyleToInternal(self._unifiedStyle) styles = self._convertUnifiedStyleToInternal(self._unifiedStyle)
for level in range(1, 7): for level in range(1, 7):

1
modules/serviceCenter/services/serviceGeneration/renderers/rendererPptx.py
View file

@ -90,7 +90,6 @@ class RendererPptx(BaseRenderer):
from pptx.dml.color import RGBColor from pptx.dml.color import RGBColor
if not style: if not style:
from modules.serviceCenter.services.serviceGeneration.styleDefaults import resolveStyle
style = resolveStyle(None) style = resolveStyle(None)
internalStyle = self._convertUnifiedStyleToInternal(style) internalStyle = self._convertUnifiedStyleToInternal(style)
styles = internalStyle styles = internalStyle

1
modules/serviceCenter/services/serviceGeneration/renderers/rendererXlsx.py
View file

@ -137,7 +137,6 @@ class RendererXlsx(BaseRenderer):
self.services.utils.debugLogToFile(f"EXCEL JSON CONTENT KEYS: {list(jsonContent.keys()) if isinstance(jsonContent, dict) else 'Not a dict'}", "EXCEL_RENDERER") self.services.utils.debugLogToFile(f"EXCEL JSON CONTENT KEYS: {list(jsonContent.keys()) if isinstance(jsonContent, dict) else 'Not a dict'}", "EXCEL_RENDERER")
if not style: if not style:
from modules.serviceCenter.services.serviceGeneration.styleDefaults import resolveStyle
style = resolveStyle(None) style = resolveStyle(None)
self._unifiedStyle = style self._unifiedStyle = style

5
modules/serviceCenter/services/serviceKnowledge/subConnectorSyncGmail.py
View file

@ -26,6 +26,7 @@ from typing import Any, Callable, Dict, List, Optional
from modules.serviceCenter.services.serviceKnowledge.subTextClean import cleanEmailBody from modules.serviceCenter.services.serviceKnowledge.subTextClean import cleanEmailBody
from modules.serviceCenter.services.serviceKnowledge.subWalkerHelpers import ( from modules.serviceCenter.services.serviceKnowledge.subWalkerHelpers import (
WalkerTimeout, WalkerTimeout,
extractWithTimeout as _extractWithTimeout,
ingestWithTimeout, ingestWithTimeout,
logItemStart, logItemStart,
) )
@ -564,10 +565,6 @@ async def _ingestAttachments(
attLabel = f"{messageId}/att:{stub['attachmentId']}/{fileName}" attLabel = f"{messageId}/att:{stub['attachmentId']}/{fileName}"
logItemStart("gmail-attachment", attLabel, sizeBytes=stub.get("size") or None, mime=mimeType) logItemStart("gmail-attachment", attLabel, sizeBytes=stub.get("size") or None, mime=mimeType)
from modules.serviceCenter.services.serviceKnowledge.subWalkerHelpers import (
extractWithTimeout as _extractWithTimeout,
)
def _runAttExtraction(): def _runAttExtraction():
return runExtraction( return runExtraction(
extractorRegistry, chunkerRegistry, extractorRegistry, chunkerRegistry,

2
modules/serviceCenter/services/serviceSubscription/mainServiceSubscription.py
View file

@ -414,7 +414,6 @@ class SubscriptionService:
mandateLabel = mandateId mandateLabel = mandateId
try: try:
from modules.datamodels.datamodelUam import Mandate
from modules.security.rootAccess import getRootDbAppConnector from modules.security.rootAccess import getRootDbAppConnector
appDb = getRootDbAppConnector() appDb = getRootDbAppConnector()
rows = appDb.getRecordset(Mandate, recordFilter={"id": mandateId}) rows = appDb.getRecordset(Mandate, recordFilter={"id": mandateId})
@ -937,7 +936,6 @@ def _buildInvoiceSummaryHtml(
) -> str: ) -> str:
"""Build an HTML invoice summary block for inclusion in the activation email.""" """Build an HTML invoice summary block for inclusion in the activation email."""
import html as htmlmod import html as htmlmod
from modules.interfaces.interfaceDbSubscription import getRootInterface as getSubRootInterface
subInterface = getSubRootInterface() subInterface = getSubRootInterface()
userCount = subInterface.countActiveUsers(mandateId) userCount = subInterface.countActiveUsers(mandateId)

2
modules/system/databaseHealth.py
View file

@ -805,7 +805,6 @@ def _discoverLegacyTables(dbFilter: Optional[str] = None) -> List[dict]:
Returns a list of dicts: {db, table, rowCount, sizeBytes}. Returns a list of dicts: {db, table, rowCount, sizeBytes}.
""" """
from modules.datamodels.datamodelBase import MODEL_REGISTRY from modules.datamodels.datamodelBase import MODEL_REGISTRY
from modules.dbHelpers.fkRegistry import ensureModelsLoaded
ensureModelsLoaded() ensureModelsLoaded()
registeredDbs = getRegisteredDatabases() registeredDbs = getRegisteredDatabases()
@ -854,7 +853,6 @@ def _dropLegacyTable(dbName: str, tableName: str) -> dict:
Raises ValueError if the table is model-backed (safety guard). Raises ValueError if the table is model-backed (safety guard).
""" """
from modules.datamodels.datamodelBase import MODEL_REGISTRY from modules.datamodels.datamodelBase import MODEL_REGISTRY
from modules.dbHelpers.fkRegistry import ensureModelsLoaded
ensureModelsLoaded() ensureModelsLoaded()
if tableName in MODEL_REGISTRY: if tableName in MODEL_REGISTRY:

1
modules/workflowAutomation/engine/executors/actionNodeExecutor.py
View file

@ -305,7 +305,6 @@ def _buildConnectionRefDict(connRef: str, chatService, services) -> Optional[Dic
def _schemaCarriesConnectionProvenance(outputSchema: str) -> bool: def _schemaCarriesConnectionProvenance(outputSchema: str) -> bool:
"""True iff the port schema declares ``carriesConnectionProvenance`` in the catalog.""" """True iff the port schema declares ``carriesConnectionProvenance`` in the catalog."""
from modules.nodeCatalog.portTypes import PORT_TYPE_CATALOG
schema = PORT_TYPE_CATALOG.get(outputSchema) schema = PORT_TYPE_CATALOG.get(outputSchema)
return bool(getattr(schema, "carriesConnectionProvenance", False)) return bool(getattr(schema, "carriesConnectionProvenance", False))

2
modules/workflowAutomation/helpers.py
View file

@ -203,7 +203,6 @@ def _validateWorkflowAccess(
if action == "execute": if action == "execute":
targetInstanceId = workflow.get("targetFeatureInstanceId") targetInstanceId = workflow.get("targetFeatureInstanceId")
if targetInstanceId: if targetInstanceId:
from modules.interfaces.interfaceDbApp import getRootInterface
access = getRootInterface().getFeatureAccess(userId, targetInstanceId) access = getRootInterface().getFeatureAccess(userId, targetInstanceId)
if access and access.get("enabled"): if access and access.get("enabled"):
return return
@ -582,7 +581,6 @@ def _getWorkflowsJoinedPaginated(
paginationParams: PaginationParams, paginationParams: PaginationParams,
) -> dict: ) -> dict:
"""SQL listing: AutoWorkflow LEFT JOIN aggregated AutoRun stats (one query + count).""" """SQL listing: AutoWorkflow LEFT JOIN aggregated AutoRun stats (one query + count)."""
from modules.connectors.connectorDbPostgre import getModelFields, parseRecordFields
wfFields = getModelFields(AutoWorkflow) wfFields = getModelFields(AutoWorkflow)
whereClause, orderClause, limitClause, values = _buildJoinedWorkflowWhereOrderLimit( whereClause, orderClause, limitClause, values = _buildJoinedWorkflowWhereOrderLimit(

1
modules/workflowAutomation/mainWorkflowAutomation.py
View file

@ -291,7 +291,6 @@ def onInstanceCreate(mandateId: str, instanceId: str, featureCode: str, template
"""Create workflow instances from template definitions when a feature instance is created.""" """Create workflow instances from template definitions when a feature instance is created."""
from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface
from modules.security.rootAccess import getRootUser from modules.security.rootAccess import getRootUser
from modules.shared.i18nRegistry import resolveText
rootUser = getRootUser() rootUser = getRootUser()
waInterface = _getWorkflowAutomationInterface(rootUser, mandateId, instanceId) waInterface = _getWorkflowAutomationInterface(rootUser, mandateId, instanceId)

2
modules/workflows/methods/methodAi/actions/process.py
View file

@ -37,7 +37,6 @@ def _action_docs_to_content_parts(services, docs: List[Any]) -> List[ContentPart
"""Extract content from ActionDocument-like objects in memory (no persistence). """Extract content from ActionDocument-like objects in memory (no persistence).
Decodes base64, runs extraction pipeline, returns ContentParts for AI. Decodes base64, runs extraction pipeline, returns ContentParts for AI.
""" """
from modules.datamodels.datamodelExtraction import ExtractionOptions, MergeStrategy
all_parts = [] all_parts = []
extraction = services.extraction extraction = services.extraction
@ -78,7 +77,6 @@ def _resolve_file_refs_to_content_parts(services, fileIdRefs) -> List[ContentPar
references, not chat message attachments. In the agent/chat context, references, not chat message attachments. In the agent/chat context,
``DocumentItemReference`` holds ChatDocument IDs that must be resolved ``DocumentItemReference`` holds ChatDocument IDs that must be resolved
via ``getChatDocumentsFromDocumentList`` instead.""" via ``getChatDocumentsFromDocumentList`` instead."""
from modules.datamodels.datamodelExtraction import ExtractionOptions, MergeStrategy
extraction = services.extraction extraction = services.extraction
if not extraction: if not extraction:

568
scripts/script_analyze_platform_module_graph.py Normal file
View file

@ -0,0 +1,568 @@
# Copyright (c) 2026 PowerOn AG
# All rights reserved.
"""
Deep platform-core module import graph analysis.
Output: local/notes/refernce-analysis/import-analysis-platform-modules.md
Usage:
python platform-core/scripts/script_analyze_platform_module_graph.py
"""
from __future__ import annotations
import ast
import os
import sys
from collections import defaultdict
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple
SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))
from script_analyze_porta_imports import ( # noqa: E402
OUTPUT_ROOT,
PLATFORM_ROOT,
SKIP_DIR_NAMES,
_collectPlatformModules,
_getPlatformContainer,
_platformModuleId,
_resolvePlatformImportTarget,
_resolvePlatformRelativeImport,
_writeText,
)
OUTPUT_FILE = OUTPUT_ROOT / "import-analysis-platform-modules.md"
LAYER_ORDER = {
"shared": 0,
"datamodels": 1,
"connectors": 2,
"nodeCatalog": 2,
"dbHelpers": 3,
"interfaces": 4,
"system": 4,
"security": 4,
"auth": 4,
"aicore": 4,
"demoConfigs": 4,
"serviceCenter": 5,
"workflows": 5,
"workflowAutomation": 5,
"features.commcoach": 5,
"features.neutralization": 5,
"features.realEstate": 5,
"features.realestate": 5,
"features.redmine": 5,
"features.teamsbot": 5,
"features.trustee": 5,
"features.workspace": 5,
"routes": 6,
"app": 7,
}
@dataclass
class ScopedImport:
target: str
rawModule: str
position: str
scope: str
isInternal: bool
isStdLib: bool
def _shortModule(moduleId: str) -> str:
parts = moduleId.replace("platform-core.", "").split(".")
if len(parts) <= 3:
return ".".join(parts)
return ".".join(parts[-3:])
LIFECYCLE_SCOPE_MARKERS = (
"lifespan",
"onBootstrap",
"onStart",
"onStop",
"onInstanceCreate",
"onMandateDelete",
"registerFeature",
"preWarm",
)
def _layerOf(moduleId: str) -> Optional[int]:
container = _getPlatformContainer(moduleId)
if container is None:
return None
return LAYER_ORDER.get(container)
def _isStdLibModule(moduleName: str) -> bool:
root = moduleName.split(".")[0]
if root.startswith("_"):
return False
if root in sys.builtin_module_names:
return True
if hasattr(sys, "stdlib_module_names") and root in sys.stdlib_module_names:
return True
return False
class _DetailedImportVisitor(ast.NodeVisitor):
def __init__(self, filePath: Path):
self.filePath = filePath
self.imports: List[ScopedImport] = []
self._scopeStack: List[str] = []
@property
def _currentScope(self) -> str:
return self._scopeStack[-1] if self._scopeStack else ""
def _position(self) -> str:
return "code" if self._scopeStack else "header"
def _add(self, rawModule: str, resolved: str, isInternal: bool) -> None:
self.imports.append(
ScopedImport(
target=resolved,
rawModule=rawModule,
position=self._position(),
scope=self._currentScope,
isInternal=isInternal,
isStdLib=_isStdLibModule(rawModule) if not isInternal else False,
)
)
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
self._scopeStack.append(f"function {node.name}")
self.generic_visit(node)
self._scopeStack.pop()
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
self._scopeStack.append(f"function {node.name}")
self.generic_visit(node)
self._scopeStack.pop()
def visit_ClassDef(self, node: ast.ClassDef) -> None:
self._scopeStack.append(f"class {node.name}")
self.generic_visit(node)
self._scopeStack.pop()
def visit_Import(self, node: ast.Import) -> None:
for alias in node.names:
resolved, isInternal = _resolvePlatformImportTarget(self.filePath, alias.name)
self._add(alias.name, resolved, isInternal)
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
if node.level > 0:
resolved = _resolvePlatformRelativeImport(self.filePath, node)
if resolved:
suffix = node.module or ""
raw = ("." * node.level) + suffix
self._add(raw, resolved, True)
return
if not node.module:
return
resolved, isInternal = _resolvePlatformImportTarget(self.filePath, node.module)
self._add(node.module, resolved, isInternal)
def _collectDetailedImports() -> Dict[str, List[ScopedImport]]:
byModule: Dict[str, List[ScopedImport]] = {}
pyFiles: List[Path] = []
appFile = PLATFORM_ROOT / "app.py"
if appFile.exists():
pyFiles.append(appFile)
for root, dirs, files in os.walk(PLATFORM_ROOT / "modules"):
dirs[:] = [d for d in dirs if d not in SKIP_DIR_NAMES]
for fileName in files:
if fileName.endswith(".py"):
pyFiles.append(Path(root) / fileName)
for filePath in pyFiles:
moduleId = _platformModuleId(filePath)
if _getPlatformContainer(moduleId) is None:
continue
try:
tree = ast.parse(filePath.read_text(encoding="utf-8"), filename=str(filePath))
except (SyntaxError, UnicodeDecodeError):
continue
visitor = _DetailedImportVisitor(filePath)
visitor.visit(tree)
byModule[moduleId] = visitor.imports
return byModule
def _internalGraph(importsByModule: Dict[str, List[ScopedImport]]) -> Dict[str, Set[str]]:
graph: Dict[str, Set[str]] = defaultdict(set)
for source, items in importsByModule.items():
for item in items:
if item.isInternal and item.target.startswith("platform-core."):
graph[source].add(item.target)
return dict(graph)
def _mutualPairs(
graph: Dict[str, Set[str]],
moduleFilter: Optional[Set[str]] = None,
) -> List[Tuple[str, str]]:
pairs: List[Tuple[str, str]] = []
seen: Set[Tuple[str, str]] = set()
sources = moduleFilter if moduleFilter is not None else set(graph.keys())
for source in sorted(sources):
for target in graph.get(source, set()):
if moduleFilter is not None and target not in moduleFilter:
continue
if target not in graph or source not in graph[target]:
continue
key = tuple(sorted((source, target)))
if key not in seen:
seen.add(key)
pairs.append(key)
return pairs
def _tarjanScc(graph: Dict[str, Set[str]]) -> List[List[str]]:
index = 0
stack: List[str] = []
onStack: Set[str] = set()
indices: Dict[str, int] = {}
lowLink: Dict[str, int] = {}
result: List[List[str]] = []
nodes = set(graph.keys())
for targets in graph.values():
nodes.update(targets)
def strongConnect(node: str) -> None:
nonlocal index
indices[node] = index
lowLink[node] = index
index += 1
stack.append(node)
onStack.add(node)
for neighbor in graph.get(node, set()):
if neighbor not in indices:
strongConnect(neighbor)
lowLink[node] = min(lowLink[node], lowLink[neighbor])
elif neighbor in onStack:
lowLink[node] = min(lowLink[node], indices[neighbor])
if lowLink[node] == indices[node]:
component: List[str] = []
while True:
w = stack.pop()
onStack.remove(w)
component.append(w)
if w == node:
break
if len(component) > 1 or (len(component) == 1 and component[0] in graph.get(component[0], set())):
result.append(sorted(component))
for node in sorted(nodes):
if node not in indices:
strongConnect(node)
return sorted(result, key=lambda c: (len(c), c[0]), reverse=True)
def _canReach(graph: Dict[str, Set[str]], start: str, goal: str, skipEdge: Optional[Tuple[str, str]] = None) -> bool:
visited: Set[str] = set()
def dfs(node: str) -> bool:
if node == goal:
return True
if node in visited:
return False
visited.add(node)
for nxt in graph.get(node, set()):
if skipEdge and node == skipEdge[0] and nxt == skipEdge[1]:
continue
if dfs(nxt):
return True
return False
return dfs(start)
def _assessMutualPair(a: str, b: str) -> str:
containerA = _getPlatformContainer(a)
containerB = _getPlatformContainer(b)
layerA = _layerOf(a)
layerB = _layerOf(b)
sameContainer = containerA == containerB
if sameContainer:
if layerA is not None and layerA >= 5:
return "Prüfen — Feature/Service-interner Gegenimport; oft Lazy-Import-Workaround, Zyklus im Container."
return "Prüfen — gegenseitiger Import im gleichen Container; meist absichtlicher Lazy-Import gegen Zyklus."
if layerA is not None and layerB is not None:
if layerA < layerB and layerB < layerA:
pass
upward = (layerA > layerB and layerB is not None) or (layerB > layerA and layerA is not None)
if upward:
return "Refactor-Kandidat — untere Schicht importiert obere und umgekehrt (Layer-Verletzung)."
return "Refactor-Kandidat — Cross-Container-Gegenimport; Layer-Grenze prüfen."
def _assessCycle(component: List[str]) -> str:
if len(component) == 1:
return "OK — Package-Reexport/Self-Import (__init__ ↔ Submodul); typisch für Barrel-Module."
containers = {_getPlatformContainer(m) for m in component}
containers.discard(None)
layers = [layer for m in component if (layer := _layerOf(m)) is not None]
if len(containers) == 1:
container = next(iter(containers))
if LAYER_ORDER.get(container or "", 99) >= 5:
return "Prüfen — Zyklus innerhalb Feature/Service-Cluster; oft bekanntes Deferred-Coupling."
return "Prüfen — Intra-Container-Loop; Lazy-Imports prüfen ob extrahierbar."
if layers and max(layers) - min(layers) >= 2:
return "Refactor-Kandidat — Loop über mehrere Layer/Container; Architektur-Grenze verletzt."
return "Prüfen — Cross-Container-Loop; Abhängigkeit entkoppeln oder Typ/Protocol extrahieren."
def _assessLazyStdLib(moduleId: str, item: ScopedImport) -> str:
heavy = {"json", "csv", "xml", "pickle", "sqlite3", "subprocess", "multiprocessing"}
root = item.rawModule.split(".")[0]
if root in heavy:
return "OK — schwere Stdlib lazy (Startup/optional)."
if "TYPE_CHECKING" in item.scope:
return "OK — typing-only Kontext."
return "Harmlos — Stdlib lazy in Code-Scope; kein Architektur-Risiko."
def _assessMovable(moduleId: str, item: ScopedImport, graph: Dict[str, Set[str]], headerTargets: Set[str]) -> str:
if item.target in headerTargets:
return "Redundant — bereits im Header importiert; Lazy-Import entfernen."
if any(marker in item.scope for marker in LIFECYCLE_SCOPE_MARKERS):
return "Beabsichtigt lazy — Startup/Lifecycle-Hook; nicht in Header verschieben."
if _canReach(graph, item.target, moduleId):
return "Muss lazy bleiben — Header-Import würde Zyklus erzeugen."
return "Verschiebbar — kann vermutlich in den Header."
def _renderMarkdown(
importsByModule: Dict[str, List[ScopedImport]],
graph: Dict[str, Set[str]],
) -> str:
modulesByContainer: Dict[str, Set[str]] = defaultdict(set)
for moduleId in importsByModule:
container = _getPlatformContainer(moduleId)
if container:
modulesByContainer[container].add(moduleId)
lines = [
"# Import-Analyse Platform — Modul-Graph",
"",
f"- **Generiert:** {date.today().isoformat()}",
"- **Script:** `platform-core/scripts/script_analyze_platform_module_graph.py`",
"- **Scope:** interne `modules.*`-Imports (inkl. lazy)",
"",
"## Legende Beurteilung",
"",
"| Stufe | Bedeutung |",
"|-------|-----------|",
"| OK / Harmlos | kein Handlungsbedarf |",
"| Verschiebbar | Lazy-Import kann vermutlich in Header |",
"| Redundant | doppelter Import (Header + Code) |",
"| Prüfen | bekannt möglich, bewusst prüfen |",
"| Beabsichtigt lazy | Startup/Lifecycle — nicht in Header |",
"| Muss lazy bleiben | Zyklusvermeidung |",
"| Refactor-Kandidat | Layer-/Architektur-Thema |",
"",
]
# --- Mutual pairs per container ---
lines.extend(["## Gegenseitige Modul-Imports (Paare)", ""])
totalPairs = 0
for container in sorted(modulesByContainer.keys()):
moduleSet = modulesByContainer[container]
pairs = _mutualPairs(graph, moduleSet)
if not pairs:
continue
totalPairs += len(pairs)
lines.append(f"### Container `{container}`")
lines.append("")
lines.append("| Modul A | Modul B | Beurteilung |")
lines.append("|---------|---------|-------------|")
for a, b in pairs:
lines.append(
f"| `{_shortModule(a)}` | `{_shortModule(b)}` | {_assessMutualPair(a, b)} |"
)
lines.append("")
crossPairs = [
p for p in _mutualPairs(graph)
if _getPlatformContainer(p[0]) != _getPlatformContainer(p[1])
]
if crossPairs:
lines.extend(["### Cross-Container (gegenseitig)", ""])
lines.append("| Modul A | Container A | Modul B | Container B | Beurteilung |")
lines.append("|---------|-------------|---------|-------------|-------------|")
for a, b in crossPairs:
lines.append(
f"| `{_shortModule(a)}` | `{_getPlatformContainer(a)}` | "
f"`{_shortModule(b)}` | `{_getPlatformContainer(b)}` | {_assessMutualPair(a, b)} |"
)
lines.append("")
if totalPairs == 0 and not crossPairs:
lines.append("_Keine gegenseitigen Modul-Paare gefunden._")
lines.append("")
# --- Cycles ---
sccList = _tarjanScc(graph)
lines.extend(["## Import-Loops (über mehrere Module)", ""])
if not sccList:
lines.append("_Keine Strongly-Connected Components (>1 Knoten) gefunden._")
lines.append("")
else:
lines.append(f"**{len(sccList)} Loop-Gruppe(n)** (Tarjan SCC, nur interne Module).")
lines.append("")
for index, component in enumerate(sccList, start=1):
containers = sorted({c for m in component if (c := _getPlatformContainer(m))})
lines.append(f"### Loop {index} — {len(component)} Module")
lines.append("")
lines.append(f"- **Container:** {', '.join(f'`{c}`' for c in containers)}")
lines.append(f"- **Beurteilung:** {_assessCycle(component)}")
lines.append("- **Module:**")
for moduleId in component:
lines.append(f" - `{moduleId}`")
if len(component) <= 8:
chainHint = " → ".join(_shortModule(m) for m in component) + f" → `{_shortModule(component[0])}`"
lines.append(f"- **Ring (Auszug):** {chainHint}")
lines.append("")
# --- Lazy stdlib ---
lines.extend(["## Lazy Stdlib-Imports (in Code-Scope)", ""])
stdlibRows: List[Tuple[str, str, str, str, str]] = []
for moduleId, items in sorted(importsByModule.items()):
for item in items:
if item.position == "code" and item.isStdLib:
stdlibRows.append(
(
_shortModule(moduleId),
_getPlatformContainer(moduleId) or "",
item.rawModule,
item.scope or "(class/function)",
_assessLazyStdLib(moduleId, item),
)
)
if stdlibRows:
lines.append("| Modul | Container | Import | Scope | Beurteilung |")
lines.append("|-------|-----------|--------|-------|-------------|")
for row in stdlibRows:
lines.append(f"| `{row[0]}` | `{row[1]}` | `{row[2]}` | {row[3]} | {row[4]} |")
lines.append("")
else:
lines.append("_Keine lazy Stdlib-Imports in Code-Scope._")
lines.append("")
# --- Lazy internal movable ---
lines.extend(["## Lazy interne Imports — Header möglich?", ""])
movableRows: List[Tuple[str, str, str, str, str]] = []
intentionalRows: List[Tuple[str, str, str, str, str]] = []
mustStayRows: List[Tuple[str, str, str, str, str]] = []
redundantRows: List[Tuple[str, str, str, str, str]] = []
for moduleId, items in sorted(importsByModule.items()):
headerTargets = {i.target for i in items if i.position == "header" and i.isInternal}
for item in items:
if item.position != "code" or not item.isInternal:
continue
verdict = _assessMovable(moduleId, item, graph, headerTargets)
row = (
_shortModule(moduleId),
_getPlatformContainer(moduleId) or "",
_shortModule(item.target),
item.scope or "(code)",
verdict,
)
if verdict.startswith("Verschiebbar"):
movableRows.append(row)
elif verdict.startswith("Beabsichtigt"):
intentionalRows.append(row)
elif verdict.startswith("Redundant"):
redundantRows.append(row)
elif verdict.startswith("Muss lazy"):
mustStayRows.append(row)
if intentionalRows:
lines.append("### Beabsichtigt lazy (Startup/Lifecycle)")
lines.append("")
lines.append(f"**{len(intentionalRows)}** Einträge — lazy in lifespan/onBootstrap/…; kein Refactor nötig.")
lines.append("")
if movableRows:
lines.append("### Verschiebbar in Header")
lines.append("")
lines.append("| Modul | Container | Import-Ziel | Scope | Beurteilung |")
lines.append("|-------|-----------|-------------|-------|-------------|")
for row in movableRows:
lines.append(f"| `{row[0]}` | `{row[1]}` | `{row[2]}` | {row[3]} | {row[4]} |")
lines.append("")
if mustStayRows:
lines.append("### Muss lazy bleiben (Zyklus)")
lines.append("")
lines.append(f"**{len(mustStayRows)}** Einträge — Auszug (max. 40):")
lines.append("")
lines.append("| Modul | Container | Import-Ziel | Scope | Beurteilung |")
lines.append("|-------|-----------|-------------|-------|-------------|")
for row in mustStayRows[:40]:
lines.append(f"| `{row[0]}` | `{row[1]}` | `{row[2]}` | {row[3]} | {row[4]} |")
if len(mustStayRows) > 40:
lines.append("")
lines.append(f"_… und {len(mustStayRows) - 40} weitere._")
lines.append("")
if redundantRows:
lines.append("### Redundant (Header + Code)")
lines.append("")
lines.append("| Modul | Container | Import-Ziel | Scope | Beurteilung |")
lines.append("|-------|-----------|-------------|-------|-------------|")
for row in redundantRows:
lines.append(f"| `{row[0]}` | `{row[1]}` | `{row[2]}` | {row[3]} | {row[4]} |")
lines.append("")
if not movableRows and not mustStayRows and not redundantRows and not intentionalRows:
lines.append("_Keine lazy internen Imports gefunden._")
lines.append("")
lines.extend(
[
"## Kurzfassung",
"",
f"- Gegenseitige Modul-Paare (intra-container): **{totalPairs}**",
f"- Gegenseitige Modul-Paare (cross-container): **{len(crossPairs)}**",
f"- Import-Loop-Gruppen (SCC): **{len(sccList)}** (davon Self-Loop: **{sum(1 for c in sccList if len(c) == 1)}**)",
f"- Lazy Stdlib-Imports: **{len(stdlibRows)}**",
f"- Lazy intern / beabsichtigt (Lifecycle): **{len(intentionalRows)}**",
f"- Lazy intern / verschiebbar: **{len(movableRows)}**",
f"- Lazy intern / Zyklus (muss bleiben): **{len(mustStayRows)}**",
f"- Lazy intern / redundant: **{len(redundantRows)}**",
"",
]
)
return "\n".join(lines)
def main() -> None:
print("Collecting detailed platform imports...")
importsByModule = _collectDetailedImports()
graph = _internalGraph(importsByModule)
print(f" modules: {len(importsByModule)}")
print(f" internal edges: {sum(len(v) for v in graph.values())}")
markdown = _renderMarkdown(importsByModule, graph)
_writeText(OUTPUT_FILE, markdown)
print(f"Written: {OUTPUT_FILE}")
if __name__ == "__main__":
main()

898
scripts/script_analyze_porta_imports.py Normal file
View file

@ -0,0 +1,898 @@
# Copyright (c) 2026 PowerOn AG
# All rights reserved.
"""
Analyze all imports (including lazy/dynamic) for PowerOn PORTA UI and platform-core.
Outputs under local/notes/refernce-analysis/:
platform/modules/*.md one file per Python module
platform/containers/*.md aggregated stats per container
platform/container-network.drawio
ui/modules/*.md
ui/containers/*.md
ui/container-network.drawio
README.md
Usage:
python platform-core/scripts/script_analyze_porta_imports.py
"""
from __future__ import annotations
import ast
import html
import math
import os
import re
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import date
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Set, Tuple
SCRIPT_DIR = Path(__file__).resolve().parent
PLATFORM_ROOT = SCRIPT_DIR.parent
REPO_ROOT = PLATFORM_ROOT.parent
UI_ROOT = REPO_ROOT / "ui-nyla"
OUTPUT_ROOT = REPO_ROOT / "local" / "notes" / "refernce-analysis"
SKIP_DIR_NAMES = {
"__pycache__",
"node_modules",
".git",
"dist",
"build",
".venv",
"venv",
".tox",
".mypy_cache",
".pytest_cache",
}
UI_SKIP_GLOBS = ("**/*.test.ts", "**/*.test.tsx", "test/**")
@dataclass
class ImportRecord:
importedModule: str
position: str # "header" | "code"
isInternal: bool
sourceContainer: Optional[str] = None
targetContainer: Optional[str] = None
@dataclass
class ModuleAnalysis:
context: str # "platform" | "ui"
moduleId: str
filePath: Path
container: str
containerPath: str
imports: List[ImportRecord] = field(default_factory=list)
def _sanitizeFileName(value: str) -> str:
return re.sub(r"[^A-Za-z0-9._-]+", "_", value)
def _writeText(path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
# ---------------------------------------------------------------------------
# Platform (Python)
# ---------------------------------------------------------------------------
def _platformModuleId(filePath: Path) -> str:
rel = filePath.relative_to(PLATFORM_ROOT)
if filePath.name == "__init__.py":
parts = rel.parent.parts
else:
parts = rel.with_suffix("").parts
return "platform-core." + ".".join(parts)
def _platformContainerPath(container: str) -> str:
if container == "app":
return "platform-core/app.py"
if container.startswith("features."):
featureCode = container.split(".", 1)[1]
return f"platform-core/modules/features/{featureCode}"
return f"platform-core/modules/{container}"
def _getPlatformContainer(moduleId: str) -> Optional[str]:
if moduleId == "platform-core.app":
return "app"
if not moduleId.startswith("platform-core."):
return None
parts = moduleId.replace("platform-core.", "").split(".")
if not parts:
return "app"
if parts[0] in ("tests", "scripts") or parts[0].startswith("script_"):
return None
if parts[0] != "modules" or len(parts) < 2:
return "app"
container = parts[1]
if container == "features" and len(parts) > 2:
return f"features.{parts[2]}"
return container
def _resolvePlatformRelativeImport(currentFile: Path, importNode: ast.ImportFrom) -> Optional[str]:
dotCount = importNode.level
moduleSuffix = importNode.module or ""
currentDir = currentFile.parent
baseDir = currentDir
for _ in range(dotCount - 1):
baseDir = baseDir.parent
if moduleSuffix:
candidate = baseDir / Path(moduleSuffix.replace(".", os.sep))
else:
candidate = baseDir
pyFile = candidate.with_suffix(".py")
if pyFile.exists():
return _platformModuleId(pyFile)
initFile = candidate / "__init__.py"
if initFile.exists():
return _platformModuleId(initFile)
rel = candidate.relative_to(PLATFORM_ROOT) if candidate.is_relative_to(PLATFORM_ROOT) else None
if rel is None:
return None
return "platform-core." + ".".join(rel.with_suffix("").parts)
def _resolvePlatformImportTarget(currentFile: Path, importedName: str) -> Tuple[str, bool]:
if importedName.startswith("."):
return importedName, False
if importedName.startswith("modules."):
parts = importedName.split(".")
checkPath = PLATFORM_ROOT
for part in parts:
checkPath = checkPath / part
if checkPath.with_suffix(".py").exists():
return _platformModuleId(checkPath.with_suffix(".py")), True
if checkPath.is_dir() and (checkPath / "__init__.py").exists():
return _platformModuleId(checkPath / "__init__.py"), True
return f"platform-core.{importedName.replace('.', '.')}", True
return importedName, False
class _PythonImportVisitor(ast.NodeVisitor):
def __init__(self, filePath: Path):
self.filePath = filePath
self.imports: List[ImportRecord] = []
self._inCodeScope = False
def _addImport(self, importedModule: str, isInternal: bool) -> None:
position = "code" if self._inCodeScope else "header"
sourceContainer = _getPlatformContainer(_platformModuleId(self.filePath))
targetContainer = _getPlatformContainer(importedModule) if isInternal else None
self.imports.append(
ImportRecord(
importedModule=importedModule,
position=position,
isInternal=isInternal,
sourceContainer=sourceContainer,
targetContainer=targetContainer,
)
)
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
previous = self._inCodeScope
self._inCodeScope = True
self.generic_visit(node)
self._inCodeScope = previous
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
previous = self._inCodeScope
self._inCodeScope = True
self.generic_visit(node)
self._inCodeScope = previous
def visit_ClassDef(self, node: ast.ClassDef) -> None:
previous = self._inCodeScope
self._inCodeScope = True
self.generic_visit(node)
self._inCodeScope = previous
def visit_Import(self, node: ast.Import) -> None:
for alias in node.names:
resolved, isInternal = _resolvePlatformImportTarget(self.filePath, alias.name)
self._addImport(resolved, isInternal)
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
if node.level > 0:
resolved = _resolvePlatformRelativeImport(self.filePath, node)
if resolved:
self._addImport(resolved, True)
else:
suffix = node.module or ""
display = ("." * node.level) + suffix
self._addImport(f"(relative-unresolved) {display}", False)
return
if not node.module:
return
resolved, isInternal = _resolvePlatformImportTarget(self.filePath, node.module)
self._addImport(resolved, isInternal)
def _analyzePythonFile(filePath: Path) -> Optional[ModuleAnalysis]:
container = _getPlatformContainer(_platformModuleId(filePath))
if container is None:
return None
try:
source = filePath.read_text(encoding="utf-8")
tree = ast.parse(source, filename=str(filePath))
except (SyntaxError, UnicodeDecodeError) as error:
print(f"WARN parse failed: {filePath}: {error}")
return None
visitor = _PythonImportVisitor(filePath)
visitor.visit(tree)
moduleId = _platformModuleId(filePath)
return ModuleAnalysis(
context="platform",
moduleId=moduleId,
filePath=filePath,
container=container,
containerPath=_platformContainerPath(container),
imports=visitor.imports,
)
def _collectPlatformModules() -> List[ModuleAnalysis]:
modules: List[ModuleAnalysis] = []
scanRoots = [PLATFORM_ROOT / "modules", PLATFORM_ROOT / "app.py"]
pyFiles: List[Path] = []
if scanRoots[1].exists():
pyFiles.append(scanRoots[1])
for root, dirs, files in os.walk(scanRoots[0]):
dirs[:] = [d for d in dirs if d not in SKIP_DIR_NAMES]
for fileName in files:
if fileName.endswith(".py"):
pyFiles.append(Path(root) / fileName)
for filePath in pyFiles:
analysis = _analyzePythonFile(filePath)
if analysis:
modules.append(analysis)
return modules
# ---------------------------------------------------------------------------
# UI (TypeScript)
# ---------------------------------------------------------------------------
TS_IMPORT_FROM_RE = re.compile(
r"""(?:^|\n)\s*(?:import|export)\s+(?:type\s+)?(?:[\w*\s{},\n\r]+?\sfrom\s+)?['"]([^'"]+)['"]""",
re.MULTILINE,
)
TS_SIDE_EFFECT_IMPORT_RE = re.compile(
r"""(?:^|\n)\s*import\s+['"]([^'"]+)['"]\s*;""",
re.MULTILINE,
)
TS_DYNAMIC_IMPORT_RE = re.compile(r"""import\s*\(\s*['"]([^'"]+)['"]\s*\)""")
def _uiModuleId(filePath: Path) -> str:
rel = filePath.relative_to(UI_ROOT / "src")
if filePath.name == "index.ts" or filePath.name == "index.tsx":
parts = rel.parent.parts
else:
parts = rel.with_suffix("").parts
return "ui-nyla.src." + ".".join(parts)
def _uiContainerPath(container: str) -> str:
if container.startswith("pages."):
suffix = container.split(".", 1)[1]
if suffix in ("admin", "basedata", "billing", "settings", "workflowAutomation"):
return f"ui-nyla/src/pages/{suffix}"
return f"ui-nyla/src/pages/views/{suffix}"
if container.startswith("components."):
suffix = container.split(".", 1)[1]
return f"ui-nyla/src/components/{suffix}"
return f"ui-nyla/src/{container}"
def _getUiContainer(moduleId: str) -> Optional[str]:
if not moduleId.startswith("ui-nyla.src."):
return None
parts = moduleId.replace("ui-nyla.src.", "").split(".")
if not parts:
return None
top = parts[0]
if top == "test":
return None
if top == "pages":
if len(parts) >= 3 and parts[1] == "views":
return f"pages.{parts[2]}"
if len(parts) >= 2:
return f"pages.{parts[1]}"
return "pages"
if top == "components" and len(parts) >= 2:
return f"components.{parts[1]}"
return top
def _resolveUiImport(currentFile: Path, spec: str) -> Tuple[str, bool]:
if spec.startswith("."):
resolvedPath = (currentFile.parent / spec).resolve()
candidates = [
resolvedPath,
resolvedPath.with_suffix(".ts"),
resolvedPath.with_suffix(".tsx"),
resolvedPath / "index.ts",
resolvedPath / "index.tsx",
]
for candidate in candidates:
if candidate.exists() and candidate.is_relative_to(UI_ROOT / "src"):
return _uiModuleId(candidate), True
relDisplay = spec
return relDisplay, False
return spec, False
def _findTsImportPosition(source: str, matchStart: int) -> str:
depth = 0
inFunction = False
functionDepth = 0
i = 0
while i < matchStart:
char = source[i]
if char == "{":
depth += 1
elif char == "}":
depth = max(0, depth - 1)
if inFunction and depth < functionDepth:
inFunction = False
i += 1
lookback = source[max(0, matchStart - 400):matchStart]
if re.search(r"(?:function\s*\w*\s*\(|=>\s*\{|(?:async\s+)?function\s+\w+\s*\()", lookback):
tail = lookback[lookback.rfind("\n") + 1:]
if "=>" in tail or "function" in tail:
bracePos = source.find("{", max(0, matchStart - 120), matchStart)
if bracePos >= 0:
return "code"
return "header" if depth == 0 and not inFunction else "code"
def _analyzeTypeScriptFile(filePath: Path) -> Optional[ModuleAnalysis]:
container = _getUiContainer(_uiModuleId(filePath))
if container is None:
return None
try:
source = filePath.read_text(encoding="utf-8")
except UnicodeDecodeError as error:
print(f"WARN read failed: {filePath}: {error}")
return None
imports: List[ImportRecord] = []
seen: Set[Tuple[str, str, str]] = set()
def _register(spec: str, position: str) -> None:
resolved, isInternal = _resolveUiImport(filePath, spec)
key = (resolved, position, spec)
if key in seen:
return
seen.add(key)
sourceContainer = container
targetContainer = _getUiContainer(resolved) if isInternal else None
imports.append(
ImportRecord(
importedModule=resolved,
position=position,
isInternal=isInternal,
sourceContainer=sourceContainer,
targetContainer=targetContainer,
)
)
for match in TS_IMPORT_FROM_RE.finditer(source):
position = _findTsImportPosition(source, match.start())
_register(match.group(1), position)
for match in TS_SIDE_EFFECT_IMPORT_RE.finditer(source):
if match.group(1) in {m.group(1) for m in TS_IMPORT_FROM_RE.finditer(source)}:
continue
position = _findTsImportPosition(source, match.start())
_register(match.group(1), position)
for match in TS_DYNAMIC_IMPORT_RE.finditer(source):
position = _findTsImportPosition(source, match.start())
_register(match.group(1), position)
moduleId = _uiModuleId(filePath)
return ModuleAnalysis(
context="ui",
moduleId=moduleId,
filePath=filePath,
container=container,
containerPath=_uiContainerPath(container),
imports=imports,
)
def _collectUiModules() -> List[ModuleAnalysis]:
srcRoot = UI_ROOT / "src"
modules: List[ModuleAnalysis] = []
for filePath in srcRoot.rglob("*"):
if not filePath.is_file():
continue
if filePath.suffix not in (".ts", ".tsx"):
continue
rel = filePath.relative_to(srcRoot).as_posix()
if rel.startswith("test/") or rel.endswith(".test.ts") or rel.endswith(".test.tsx"):
continue
analysis = _analyzeTypeScriptFile(filePath)
if analysis:
modules.append(analysis)
return modules
# ---------------------------------------------------------------------------
# Markdown output
# ---------------------------------------------------------------------------
def _renderModuleMarkdown(module: ModuleAnalysis) -> str:
lines = [
f"# Module Import Analysis: `{module.moduleId}`",
"",
f"- **Kontext:** {module.context}",
f"- **Container:** `{module.container}`",
f"- **Container-Pfad:** `{module.containerPath}`",
f"- **Datei:** `{module.filePath.relative_to(REPO_ROOT).as_posix()}`",
f"- **Import-Anzahl:** {len(module.imports)}",
"",
"## Imports",
"",
"| Modul | Position | Intern |",
"|-------|----------|--------|",
]
for item in sorted(module.imports, key=lambda x: (x.importedModule, x.position)):
internal = "ja" if item.isInternal else "nein"
lines.append(f"| `{item.importedModule}` | {item.position} | {internal} |")
if not module.imports:
lines.append("| _keine_ | | |")
lines.append("")
return "\n".join(lines)
@dataclass
class ContainerStats:
container: str
containerPath: str
importsFrom: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
exportedTo: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
mixedWith: Dict[str, Tuple[int, int]] = field(default_factory=dict)
def _buildContainerStats(modules: Iterable[ModuleAnalysis]) -> Dict[str, ContainerStats]:
statsByContainer: Dict[str, ContainerStats] = {}
for module in modules:
if module.container not in statsByContainer:
statsByContainer[module.container] = ContainerStats(
container=module.container,
containerPath=module.containerPath,
)
for item in module.imports:
if not item.isInternal:
continue
if not item.sourceContainer or not item.targetContainer:
continue
if item.sourceContainer == item.targetContainer:
continue
stats = statsByContainer[item.sourceContainer]
stats.importsFrom[item.targetContainer] += 1
targetStats = statsByContainer.get(item.targetContainer)
if targetStats is None:
targetStats = ContainerStats(
container=item.targetContainer,
containerPath=_platformContainerPath(item.targetContainer)
if module.context == "platform"
else _uiContainerPath(item.targetContainer),
)
statsByContainer[item.targetContainer] = targetStats
targetStats.exportedTo[item.sourceContainer] += 1
for containerName, stats in statsByContainer.items():
mixed: Dict[str, Tuple[int, int]] = {}
for other, outCount in stats.importsFrom.items():
inCount = stats.exportedTo.get(other, 0)
if inCount > 0:
mixed[other] = (outCount, inCount)
stats.mixedWith = mixed
return statsByContainer
def _renderContainerMarkdown(context: str, stats: ContainerStats) -> str:
importsTotal = sum(stats.importsFrom.values())
exportsTotal = sum(stats.exportedTo.values())
mixedTotal = sum(min(pair[0], pair[1]) for pair in stats.mixedWith.values())
lines = [
f"# Container Import Analysis: `{stats.container}`",
"",
f"- **Kontext:** {context}",
f"- **Container-Pfad:** `{stats.containerPath}`",
"",
"## Imports aus anderen Containern",
"",
f"- **Anzahl:** {importsTotal}",
f"- **Container ({len(stats.importsFrom)}):** "
+ (", ".join(f"`{name}` ({count})" for name, count in sorted(stats.importsFrom.items())) or "_keine_"),
"",
"## Exports zu anderen Containern",
"",
f"- **Anzahl:** {exportsTotal}",
f"- **Container ({len(stats.exportedTo)}):** "
+ (", ".join(f"`{name}` ({count})" for name, count in sorted(stats.exportedTo.items())) or "_keine_"),
"",
"## Cross (mixed Import/Export)",
"",
f"- **Anzahl bidirektionaler Paare:** {len(stats.mixedWith)}",
f"- **Mindest-Wechselzahl (min je Richtung):** {mixedTotal}",
]
if stats.mixedWith:
lines.extend(["", "| Container | Importe hinaus | Importe herein |", "|-----------|----------------|----------------|"])
for other, (outCount, inCount) in sorted(stats.mixedWith.items()):
lines.append(f"| `{other}` | {outCount} | {inCount} |")
else:
lines.append("- **Container:** _keine_")
lines.append("")
return "\n".join(lines)
def _renderReadme(platformModules: int, uiModules: int, platformContainers: int, uiContainers: int) -> str:
return f"""# PORTA Import-Analyse
Generiert am {date.today().isoformat()} durch `platform-core/scripts/script_analyze_porta_imports.py`.
## Umfang
| Kontext | Module | Container |
|---------|--------|-----------|
| platform | {platformModules} | {platformContainers} |
| ui | {uiModules} | {uiContainers} |
## Struktur
- `import-analysis-platform.md` — konsolidierte Platform-Übersicht (Tabelle)
- `import-analysis-platform-modules.md` — Modul-Graph: Gegenimporte, Loops, lazy Imports
- `import-analysis-ui.md` — konsolidierte UI-Übersicht (Tabelle)
- `platform/modules/` — ein Markdown pro Python-Modul (alle Imports inkl. lazy)
- `platform/containers/` — aggregierte Container-Statistik
- `platform/container-network.drawio` — Container-Vernetzung (schwarz=einweg, rot=mixed)
- `ui/modules/` — ein Markdown pro TS/TSX-Modul
- `ui/containers/` — aggregierte Container-Statistik
- `ui/container-network.drawio` — Container-Vernetzung
- `container-network.drawio` — kombiniert (2 Diagramm-Tabs: platform + ui)
## Position
- `header` — Import auf Modulebene (Top-Level)
- `code` — Import innerhalb von Funktion/Klasse oder dynamisch (`import()`)
## Regenerieren
```bash
python platform-core/scripts/script_analyze_porta_imports.py
```
"""
# ---------------------------------------------------------------------------
# draw.io
# ---------------------------------------------------------------------------
CONTAINER_COLORS = {
"app": "#dae8fc",
"aicore": "#d5e8d4",
"auth": "#ffe6cc",
"connectors": "#e1d5e7",
"datamodels": "#fff2cc",
"interfaces": "#f8cecc",
"routes": "#d0cee2",
"security": "#fad7ac",
"serviceCenter": "#b1ddf0",
"shared": "#f0fff0",
"workflows": "#f5f5f5",
"workflowAutomation": "#e6d0de",
"system": "#cce5ff",
"dbHelpers": "#fff0f5",
"nodeCatalog": "#f5fffa",
"pages": "#dae8fc",
"components": "#d5e8d4",
"hooks": "#ffe6cc",
"contexts": "#e1d5e7",
"api": "#fff2cc",
"layouts": "#f8cecc",
"providers": "#d0cee2",
"config": "#fad7ac",
"utils": "#b1ddf0",
"types": "#f0fff0",
"locales": "#f5f5f5",
"stores": "#e2efda",
"styles": "#fce5cd",
}
def _aggregateContainerEdges(statsByContainer: Dict[str, ContainerStats]) -> Dict[Tuple[str, str], Tuple[int, int, bool]]:
pairCounts: Dict[Tuple[str, str], Tuple[int, int]] = {}
for stats in statsByContainer.values():
for target, count in stats.importsFrom.items():
key = (stats.container, target)
outCount, inCount = pairCounts.get(key, (0, 0))
pairCounts[key] = (outCount + count, inCount)
edges: Dict[Tuple[str, str], Tuple[int, int, bool]] = {}
processed: Set[Tuple[str, str]] = set()
for (source, target), (forward, _) in list(pairCounts.items()):
pairKey = tuple(sorted((source, target)))
if pairKey in processed:
continue
processed.add(pairKey)
a, b = pairKey
aToB = pairCounts.get((a, b), (0, 0))[0]
bToA = pairCounts.get((b, a), (0, 0))[0]
if aToB == 0 and bToA == 0:
continue
if aToB > 0 and bToA > 0:
edges[(a, b)] = (aToB, bToA, True)
elif aToB > 0:
edges[(a, b)] = (aToB, 0, False)
else:
edges[(b, a)] = (bToA, 0, False)
return edges
def _generateDrawio(context: str, statsByContainer: Dict[str, ContainerStats]) -> str:
containers = sorted(statsByContainer.keys())
edges = _aggregateContainerEdges(statsByContainer)
centerX = 700
centerY = 550
radius = 430
nodeWidth = 170
nodeHeight = 62
containerPositions: Dict[str, Tuple[int, int]] = {}
for index, container in enumerate(containers):
angle = (2 * math.pi * index / max(len(containers), 1)) - math.pi / 2
x = int(centerX + radius * math.cos(angle) - nodeWidth / 2)
y = int(centerY + radius * math.sin(angle) - nodeHeight / 2)
containerPositions[container] = (x, y)
cells: List[str] = []
for container in containers:
x, y = containerPositions[container]
base = container.split(".")[0]
color = CONTAINER_COLORS.get(base, "#ffffff")
label = f"{container}\\n({sum(statsByContainer[container].importsFrom.values())} out / "
label += f"{sum(statsByContainer[container].exportedTo.values())} in)"
cellId = f"container_{container.replace('.', '_')}"
cells.append(
f""" <mxCell id="{cellId}" value="{html.escape(label)}" """
f"""style="rounded=1;whiteSpace=wrap;html=1;fillColor={color};strokeColor=#666666;fontStyle=1;fontSize=11;" """
f"""vertex="1" parent="1">
<mxGeometry x="{x}" y="{y}" width="{nodeWidth}" height="{nodeHeight}" as="geometry" />
</mxCell>"""
)
edgeId = 1000
for (source, target), (forward, backward, isMixed) in sorted(edges.items(), key=lambda item: -(item[1][0] + item[1][1])):
sourceId = f"container_{source.replace('.', '_')}"
targetId = f"container_{target.replace('.', '_')}"
if isMixed:
label = f"{forward} / {backward}"
strokeColor = "#CC0000"
else:
label = str(forward)
strokeColor = "#000000"
strokeWidth = min(1 + (forward + backward) // 15, 6)
cells.append(
f""" <mxCell id="edge_{edgeId}" value="{html.escape(label)}" """
f"""style="edgeStyle=none;rounded=0;orthogonalLoop=1;jettySize=auto;html=1;"""
f"""endArrow=block;endFill=1;strokeWidth={strokeWidth};strokeColor={strokeColor};"""
f"""fontSize=10;labelBackgroundColor=#ffffff;fontStyle=1;" """
f"""edge="1" parent="1" source="{sourceId}" target="{targetId}">
<mxGeometry relative="1" as="geometry" />
</mxCell>"""
)
edgeId += 1
innerXml = f""" <mxGraphModel dx="1434" dy="780" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="1600" pageHeight="1200" math="0" shadow="0">
<root>
<mxCell id="0" />
<mxCell id="1" parent="0" />
{chr(10).join(cells)}
</root>
</mxGraphModel>"""
return _wrapDrawioDiagram(context, innerXml)
def _wrapDrawioDiagram(context: str, innerXml: str) -> str:
return f"""<?xml version="1.0" encoding="UTF-8"?>
<mxfile host="app.diagrams.net" modified="{date.today().isoformat()}T00:00:00.000Z" agent="script_analyze_porta_imports.py" version="21.0.0" type="device">
<diagram id="{context}-container-network" name="{context} container imports">
{innerXml}
</diagram>
</mxfile>
"""
def _extractDrawioDiagramBody(drawioXml: str) -> str:
start = drawioXml.index("<mxGraphModel")
end = drawioXml.index("</diagram>")
return drawioXml[start:end]
def _combineDrawioFiles(platformDrawio: str, uiDrawio: str) -> str:
return f"""<?xml version="1.0" encoding="UTF-8"?>
<mxfile host="app.diagrams.net" modified="{date.today().isoformat()}T00:00:00.000Z" agent="script_analyze_porta_imports.py" version="21.0.0" type="device">
<diagram id="platform-container-network" name="platform container imports">
{_extractDrawioDiagramBody(platformDrawio)}
</diagram>
<diagram id="ui-container-network" name="ui container imports">
{_extractDrawioDiagramBody(uiDrawio)}
</diagram>
</mxfile>
"""
SUMMARY_FILE_PLATFORM = "import-analysis-platform.md"
SUMMARY_FILE_UI = "import-analysis-ui.md"
def _renderConsolidatedSummary(
title: str,
context: str,
detailFolder: str,
statsByContainer: Dict[str, ContainerStats],
diagramPath: str,
) -> str:
lines = [
f"# {title}",
"",
f"- **Kontext:** {context}",
f"- **Generiert:** {date.today().isoformat()}",
f"- **Detail-Dateien:** `{detailFolder}/`",
"",
"## Container",
"",
"| Container | Imports out | Exports in | Mixed | Detail |",
"|-----------|------------:|-----------:|------:|--------|",
]
for containerName in sorted(statsByContainer.keys()):
stats = statsByContainer[containerName]
importsOut = sum(stats.importsFrom.values())
exportsIn = sum(stats.exportedTo.values())
mixedCount = len(stats.mixedWith)
detailLink = f"[Detail]({detailFolder}/containers/{_sanitizeFileName(containerName)}.md)"
lines.append(
f"| `{containerName}` | {importsOut} | {exportsIn} | {mixedCount} | {detailLink} |"
)
lines.extend(
[
"",
f"Diagramm: [{diagramPath}]({diagramPath})",
"",
]
)
return "\n".join(lines)
def _writeContextOutput(context: str, modules: List[ModuleAnalysis]) -> Tuple[int, str]:
contextRoot = OUTPUT_ROOT / context
modulesDir = contextRoot / "modules"
containersDir = contextRoot / "containers"
for module in modules:
fileName = _sanitizeFileName(module.moduleId) + ".md"
_writeText(modulesDir / fileName, _renderModuleMarkdown(module))
statsByContainer = _buildContainerStats(modules)
for containerName, stats in sorted(statsByContainer.items()):
fileName = _sanitizeFileName(containerName) + ".md"
_writeText(containersDir / fileName, _renderContainerMarkdown(context, stats))
drawio = _generateDrawio(context, statsByContainer)
_writeText(contextRoot / "container-network.drawio", drawio)
return len(statsByContainer), drawio
def main() -> None:
print("Analyzing platform-core (Python)...")
platformModules = _collectPlatformModules()
print(f" modules: {len(platformModules)}")
print("Analyzing ui-nyla (TypeScript)...")
uiModules = _collectUiModules()
print(f" modules: {len(uiModules)}")
platformContainerCount, platformDrawio = _writeContextOutput("platform", platformModules)
uiContainerCount, uiDrawio = _writeContextOutput("ui", uiModules)
combinedDrawio = _combineDrawioFiles(platformDrawio, uiDrawio)
_writeText(OUTPUT_ROOT / "container-network.drawio", combinedDrawio)
readme = _renderReadme(
platformModules=len(platformModules),
uiModules=len(uiModules),
platformContainers=platformContainerCount,
uiContainers=uiContainerCount,
)
_writeText(OUTPUT_ROOT / "README.md", readme)
platformStats = _buildContainerStats(platformModules)
platformSummary = _renderConsolidatedSummary(
title="Import-Analyse Platform Core",
context="platform",
detailFolder="platform",
statsByContainer=platformStats,
diagramPath="platform/container-network.drawio",
)
_writeText(OUTPUT_ROOT / SUMMARY_FILE_PLATFORM, platformSummary)
uiStats = _buildContainerStats(uiModules)
uiSummary = _renderConsolidatedSummary(
title="Import-Analyse UI Nyla",
context="ui",
detailFolder="ui",
statsByContainer=uiStats,
diagramPath="ui/container-network.drawio",
)
_writeText(OUTPUT_ROOT / SUMMARY_FILE_UI, uiSummary)
print(f"\nOutput written to: {OUTPUT_ROOT}")
print(f" platform containers: {platformContainerCount}")
print(f" ui containers: {uiContainerCount}")
if __name__ == "__main__":
main()

165
scripts/script_remove_redundant_platform_imports.py Normal file
View file

@ -0,0 +1,165 @@
# Copyright (c) 2026 PowerOn AG
# All rights reserved.
"""
Remove redundant lazy imports in platform-core when the same internal module
is already imported at module header level.
Usage:
python platform-core/scripts/script_remove_redundant_platform_imports.py
python platform-core/scripts/script_remove_redundant_platform_imports.py --dry-run
"""
from __future__ import annotations
import argparse
import ast
import os
import sys
from pathlib import Path
from typing import Dict, List, Set, Tuple
SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))
from script_analyze_porta_imports import ( # noqa: E402
PLATFORM_ROOT,
SKIP_DIR_NAMES,
_getPlatformContainer,
_platformModuleId,
_resolvePlatformImportTarget,
_resolvePlatformRelativeImport,
)
class _RedundantImportFinder(ast.NodeVisitor):
def __init__(self, filePath: Path):
self.filePath = filePath
self.headerTargets: Set[str] = set()
self.linesToRemove: Set[int] = set()
self._scopeDepth = 0
def _resolveImportNode(self, node: ast.Import | ast.ImportFrom) -> List[Tuple[str, bool]]:
resolved: List[Tuple[str, bool]] = []
if isinstance(node, ast.ImportFrom):
if node.level > 0:
target = _resolvePlatformRelativeImport(self.filePath, node)
if target:
resolved.append((target, True))
return resolved
if not node.module:
return resolved
target, isInternal = _resolvePlatformImportTarget(self.filePath, node.module)
resolved.append((target, isInternal))
return resolved
for alias in node.names:
target, isInternal = _resolvePlatformImportTarget(self.filePath, alias.name)
resolved.append((target, isInternal))
return resolved
def _handleImportNode(self, node: ast.Import | ast.ImportFrom) -> None:
for target, isInternal in self._resolveImportNode(node):
if not isInternal or not target.startswith("platform-core."):
continue
if self._scopeDepth == 0:
self.headerTargets.add(target)
elif target in self.headerTargets:
endLine = getattr(node, "end_lineno", None) or node.lineno
for lineNo in range(node.lineno, endLine + 1):
self.linesToRemove.add(lineNo)
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
self._scopeDepth += 1
self.generic_visit(node)
self._scopeDepth -= 1
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
self._scopeDepth += 1
self.generic_visit(node)
self._scopeDepth -= 1
def visit_ClassDef(self, node: ast.ClassDef) -> None:
self._scopeDepth += 1
self.generic_visit(node)
self._scopeDepth -= 1
def visit_Import(self, node: ast.Import) -> None:
self._handleImportNode(node)
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
self._handleImportNode(node)
def _moduleIdToFilePath(moduleId: str) -> Path:
rel = moduleId.replace("platform-core.", "")
parts = rel.split(".")
candidate = PLATFORM_ROOT.joinpath(*parts).with_suffix(".py")
if candidate.exists():
return candidate
initFile = PLATFORM_ROOT.joinpath(*parts, "__init__.py")
return initFile
def _collectPythonFiles() -> List[Path]:
pyFiles: List[Path] = []
appFile = PLATFORM_ROOT / "app.py"
if appFile.exists():
pyFiles.append(appFile)
for root, dirs, files in os.walk(PLATFORM_ROOT / "modules"):
dirs[:] = [d for d in dirs if d not in SKIP_DIR_NAMES]
for fileName in files:
if fileName.endswith(".py"):
pyFiles.append(Path(root) / fileName)
return pyFiles
def _removeLines(filePath: Path, linesToRemove: Set[int], dryRun: bool) -> int:
if not linesToRemove:
return 0
lines = filePath.read_text(encoding="utf-8").splitlines(keepends=True)
newLines = [line for index, line in enumerate(lines, start=1) if index not in linesToRemove]
if not dryRun:
filePath.write_text("".join(newLines), encoding="utf-8")
return len(linesToRemove)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args()
totalRemoved = 0
filesChanged = 0
details: List[Tuple[str, int]] = []
for filePath in _collectPythonFiles():
moduleId = _platformModuleId(filePath)
if _getPlatformContainer(moduleId) is None:
continue
try:
tree = ast.parse(filePath.read_text(encoding="utf-8"), filename=str(filePath))
except (SyntaxError, UnicodeDecodeError) as error:
print(f"WARN skip {filePath}: {error}")
continue
finder = _RedundantImportFinder(filePath)
finder.visit(tree)
if not finder.linesToRemove:
continue
removed = _removeLines(filePath, finder.linesToRemove, args.dry_run)
totalRemoved += removed
filesChanged += 1
rel = filePath.relative_to(PLATFORM_ROOT.parent).as_posix()
details.append((rel, removed))
action = "would remove" if args.dry_run else "removed"
print(f"{action} {removed} from {rel}")
print(f"\nFiles touched: {filesChanged}")
print(f"Import lines {('would be ' if args.dry_run else '')}removed: {totalRemoved}")
if __name__ == "__main__":
main()