ingest command new version with unique identifir in notification message
This commit is contained in:
+583
-750
File diff suppressed because it is too large
Load Diff
+23
-16
@@ -1,17 +1,18 @@
|
||||
from ftplib import FTP, error_perm
|
||||
import os
|
||||
from pydantic import BaseModel, Field
|
||||
from typing import Optional, List
|
||||
from typing import Optional, List, Dict
|
||||
from datetime import datetime
|
||||
from time import sleep
|
||||
from paho.mqtt import client as mqtt_client
|
||||
from logging import Logger
|
||||
from uuid import uuid5, NAMESPACE_OID
|
||||
|
||||
class FTPDirInfo(BaseModel):
|
||||
name: str
|
||||
modified_time: datetime = Field(default_factory=datetime.now)
|
||||
unique: str = Field(default_factory=lambda: str(datetime.now().timestamp()))
|
||||
|
||||
|
||||
type DirDescriptors = Dict[str, FTPDirInfo]
|
||||
class FTPServer(BaseModel):
|
||||
name: str
|
||||
host: str
|
||||
@@ -20,7 +21,7 @@ class FTPServer(BaseModel):
|
||||
password: Optional[str] = None
|
||||
cwd: Optional[str] = None
|
||||
timeout: int = 15
|
||||
dirs: List[FTPDirInfo] = Field(default_factory=list)
|
||||
dirs: DirDescriptors = Field(default_factory=dict)
|
||||
notify_filters: List[str] = Field(default_factory=list)
|
||||
|
||||
def connect(self) -> FTP:
|
||||
@@ -40,8 +41,12 @@ class FTPServers(BaseModel):
|
||||
class NotificationMessage(BaseModel):
|
||||
server: str
|
||||
name: str
|
||||
unique: str
|
||||
modified_time: datetime
|
||||
|
||||
class NotificationResponse(BaseModel):
|
||||
unique: str
|
||||
|
||||
def load_ftp_servers_from_json(file_path: str) -> FTPServers:
|
||||
"""Loads FTP server configurations from a JSON file."""
|
||||
with open(file_path, 'r') as f:
|
||||
@@ -75,12 +80,12 @@ def initialize_ftp_servers(file_path: str) -> FTPServers:
|
||||
save_ftp_servers_to_json(dcp_servers, file_path)
|
||||
return dcp_servers
|
||||
|
||||
def list_new_directories(ftp_server: FTPServer) -> List[FTPDirInfo]:
|
||||
def list_new_directories(ftp_server: FTPServer) -> DirDescriptors:
|
||||
"""Lists new directories on the FTP server that are not already known."""
|
||||
with ftp_server.connect() as ftp_connection:
|
||||
if ftp_server.cwd:
|
||||
ftp_connection.cwd(ftp_server.cwd)
|
||||
new_dirs: List[FTPDirInfo] = []
|
||||
new_dirs: DirDescriptors = {}
|
||||
try:
|
||||
# main method, retrieve directories with details
|
||||
for d in ftp_connection.mlsd(): # List directory contents
|
||||
@@ -89,16 +94,17 @@ def list_new_directories(ftp_server: FTPServer) -> List[FTPDirInfo]:
|
||||
modified_time_str: str = dir_info.get('modify', '')
|
||||
modified_time: datetime = datetime.strptime(modified_time_str, '%Y%m%d%H%M%S') if modified_time_str else datetime.now()
|
||||
unique: str = dir_info.get('unique', str(datetime.now().timestamp()))
|
||||
new_dirs.append(FTPDirInfo(name=dir_name, modified_time=modified_time, unique=unique))
|
||||
new_dirs[unique]=FTPDirInfo(name=dir_name,
|
||||
modified_time=modified_time)
|
||||
except error_perm:
|
||||
# revert to simple directory listing mode
|
||||
for d in ftp_connection.nlst():
|
||||
dir_name = d
|
||||
new_dirs.append(FTPDirInfo(name=dir_name))
|
||||
|
||||
new_dirs.sort(key=lambda x: x.name, reverse=False) # Sort directories by name
|
||||
unique = uuid5(namespace=NAMESPACE_OID,name=dir_name).hex.upper()[:11]
|
||||
new_dirs[unique]=FTPDirInfo(name=dir_name)
|
||||
|
||||
|
||||
diff_dirs: List[FTPDirInfo] = list(filter(lambda x: x.name not in [d.name for d in ftp_server.dirs], new_dirs))
|
||||
diff_dirs: DirDescriptors = { k:v for k,v in new_dirs.items() if k not in ftp_server.dirs }
|
||||
ftp_server.dirs = new_dirs # Update the server's directory list with the new directories
|
||||
return diff_dirs # Return both new directories and the difference
|
||||
|
||||
@@ -113,21 +119,22 @@ def check_new_titles(mqtt: mqtt_client.Client, topic: str, file_path: str, logge
|
||||
|
||||
for server in dcp_servers.servers:
|
||||
# find differences between register and FTP server
|
||||
diff_dirs: List[FTPDirInfo] = list_new_directories(server)
|
||||
diff_dirs: DirDescriptors = list_new_directories(server)
|
||||
# do not publish complete list with garbage
|
||||
if first_run:
|
||||
logger.info("DCP first check on %s, %d DCPs found", server.name, len(diff_dirs))
|
||||
continue
|
||||
for dir in diff_dirs:
|
||||
logger.info(f"New DCP found on {server.name}: {dir.name} (Modified: {dir.modified_time}, Unique: {dir.unique})")
|
||||
for unique, dir in diff_dirs.items():
|
||||
logger.info(f"New DCP found on {server.name}: {dir.name} (Modified: {dir.modified_time}, Unique: {unique})")
|
||||
for filter_str in server.notify_filters:
|
||||
if filter_str in dir.name:
|
||||
notification = NotificationMessage(
|
||||
server=server.name,
|
||||
name=dir.name,
|
||||
modified_time=dir.modified_time
|
||||
modified_time=dir.modified_time,
|
||||
unique=unique
|
||||
)
|
||||
mqtt.publish(topic, notification.model_dump_json(), qos=2)
|
||||
mqtt.publish(topic, notification.model_dump_json(), qos=1)
|
||||
sleep(1) # Small delay to ensure message is sent
|
||||
|
||||
# Save server information with updated DCP list
|
||||
|
||||
+10
-7
@@ -15,7 +15,7 @@ from paho.mqtt.enums import CallbackAPIVersion
|
||||
from requests import Timeout, ConnectionError
|
||||
from pyutils.utils import SignalHandler
|
||||
from homeassistant import HAdevice, HAsensor, HAbutton
|
||||
from ftpexplorer import check_new_titles, FTPServers, FTPServer, NotificationMessage
|
||||
from ftpexplorer import check_new_titles, FTPServers, FTPServer, NotificationResponse
|
||||
|
||||
T = TypeVar("T", bound=SMSMessageWrapper)
|
||||
|
||||
@@ -52,17 +52,19 @@ def on_command(client: mqtt_client.Client, userdata: Any, msg: mqtt_client.MQTTM
|
||||
params: str = content["params"]
|
||||
match cmd:
|
||||
case "ingest":
|
||||
notify: NotificationMessage = NotificationMessage.model_validate(params)
|
||||
notify: NotificationResponse = NotificationResponse.model_validate(params)
|
||||
with open(DCP_SERVER_FILE, 'r') as fd:
|
||||
dcp_servers: FTPServers = FTPServers.model_validate_json(fd.read().encode('utf-8'))
|
||||
this_server: FTPServer = list(filter(lambda s: s.name == notify.server, dcp_servers.servers))[0]
|
||||
unique: str = notify.unique
|
||||
this_server: FTPServer = list(filter(lambda s: unique in s.dirs.keys(), dcp_servers.servers))[0]
|
||||
user_pswd: str = f"{this_server.username}:{this_server.password}" if this_server.username else "anonymous:"
|
||||
dir_name: str = f"{this_server.cwd}{"/" if this_server.cwd else ""}{this_server.dirs[unique].name}"
|
||||
ingest_command: AddImportJob = AddImportJob()
|
||||
assert isinstance(ingest_command.MessageBody, CommandBodyWrapper)
|
||||
ingest_command.MessageBody.Command.ParameterList.Parameter.insert(0,
|
||||
ParameterWrapper(
|
||||
Name="Folder",
|
||||
Value=f"ftp://{user_pswd}@{this_server.host}:{this_server.port}/{notify.name}"
|
||||
Value=f"ftp://{user_pswd}@{this_server.host}:{this_server.port}/{dir_name}"
|
||||
)
|
||||
)
|
||||
rv: Dict = projector.post(
|
||||
@@ -77,9 +79,9 @@ def on_command(client: mqtt_client.Client, userdata: Any, msg: mqtt_client.MQTTM
|
||||
LOGGER.warning("Invalid error response: [%s]", rv)
|
||||
return
|
||||
if error_response.MessageBody.ErrorResponse.Code != 100:
|
||||
LOGGER.error("Unable to ingest Title: [%s]", notify.name)
|
||||
LOGGER.error("Unable to ingest Title: [%s]", dir_name)
|
||||
else:
|
||||
LOGGER.info("Started ingest of Title: [%s] from [%s]", notify.name, notify.server)
|
||||
LOGGER.info("Started ingest of Title: [%s] from [%s]", dir_name, this_server.name)
|
||||
|
||||
except json.JSONDecodeError as de:
|
||||
LOGGER.error("Unable to decode command json: [%s]", de)
|
||||
@@ -353,7 +355,8 @@ def main() -> int:
|
||||
payload=sts.model_dump_json(
|
||||
indent=1,
|
||||
exclude={"MessageHeader"},
|
||||
exclude_unset=True))
|
||||
exclude_unset=True),
|
||||
qos=1)
|
||||
last_cls = sens.request
|
||||
except (Timeout, ConnectionError) as ce:
|
||||
LOGGER.debug("Projector is probably shutdown: [%s]", ce)
|
||||
|
||||
Reference in New Issue
Block a user