init
This commit is contained in:
76
CSV.py
76
CSV.py
@@ -1,4 +1,4 @@
|
||||
import json, csv, os, time, stomp
|
||||
import json, csv, os, time, stomp, logging
|
||||
from datetime import datetime
|
||||
from flask_backend.Stomp_client import get_connection
|
||||
|
||||
@@ -6,31 +6,81 @@ QUEUE_NAME = "/queue/testQueue"
|
||||
OUTPUT_DIR = "/app/output"
|
||||
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
||||
|
||||
# 🔧 Logger setup
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(levelname)s:%(name)s:%(message)s"
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class MyListener(stomp.ConnectionListener):
|
||||
def on_message(self, frame):
|
||||
logger.info(f"Přijatá zpráva z activeMQ: {frame.body}")
|
||||
|
||||
try:
|
||||
data = json.loads(frame.body)
|
||||
except json.JSONDecodeError:
|
||||
print("Received invalid JSON:", frame.body)
|
||||
except Exception as e:
|
||||
logger.error(f"Error when creating JSON: {e}\nInput was: {frame}")
|
||||
return
|
||||
|
||||
# Vytvoření souboru
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
filename = os.path.join(OUTPUT_DIR, f"message_{timestamp}.csv")
|
||||
filename = os.path.join(OUTPUT_DIR, f"actors_{timestamp}.csv")
|
||||
|
||||
try:
|
||||
with open(filename, "w", newline="") as f:
|
||||
writer = csv.writer(f)
|
||||
writer.writerow(data.keys())
|
||||
writer.writerow(data.values())
|
||||
print(f"Saved CSV: {filename}")
|
||||
"""try:
|
||||
with open(filename, "w", newline="", encoding="utf-8") as f:
|
||||
writer = csv.writer(f, delimiter=";")
|
||||
|
||||
for actor in data:
|
||||
row = ["list"]
|
||||
|
||||
for key, value in actor.items():
|
||||
value = actor.get(key, "")
|
||||
|
||||
if isinstance(value, list):
|
||||
for item in value:
|
||||
if isinstance(item, dict):
|
||||
row.extend(item.values())
|
||||
else:
|
||||
row.append(str(item))
|
||||
|
||||
|
||||
elif isinstance(value, bool):
|
||||
value = str(value).lower()
|
||||
|
||||
else:
|
||||
row.append(value)
|
||||
|
||||
writer.writerow(row)
|
||||
|
||||
logger.info(f"CSV uložen: {filename} ({len(data)} řádků)")
|
||||
except Exception as e:
|
||||
print("Error writing CSV:", e)
|
||||
logger.error(f"Chyba při zápisu CSV: {e}")"""
|
||||
|
||||
try:
|
||||
with open(filename, "w", newline="", encoding="utf-8") as f:
|
||||
writer = csv.writer(f, delimiter=";")
|
||||
|
||||
# Připojení k ActiveMQ s listenerem
|
||||
for actor in data:
|
||||
row = ["list"]
|
||||
for key, value in actor.items():
|
||||
if isinstance(value, dict): # rozbalíme dict (např. address)
|
||||
row.extend(value.values())
|
||||
else:
|
||||
row.append(value)
|
||||
writer.writerow(row)
|
||||
|
||||
logger.info(f"CSV uložen: {filename} ({len(data)} řádků)")
|
||||
except Exception as e:
|
||||
logger.error(f"Chyba při zápisu CSV: {e}")
|
||||
|
||||
# ActiveMQ client
|
||||
listener = MyListener()
|
||||
conn = get_connection(listener=listener)
|
||||
conn.subscribe(destination=QUEUE_NAME, id=1, ack="auto")
|
||||
|
||||
# keep process alive
|
||||
logger.info(f"ActiveMQ is running\nAddress: {QUEUE_NAME}")
|
||||
|
||||
# Keep the process active
|
||||
while True:
|
||||
time.sleep(1)
|
||||
|
||||
Reference in New Issue
Block a user