90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
import base64
|
|
import requests
|
|
import os
|
|
from dateutil import parser
|
|
from datetime import datetime, timezone
|
|
import logging
|
|
|
|
from config import GITEA_URL, GITEA_TOKEN, ORGANISATION, DEPOT_FICHES, DEPOT_CODE, ENV, ENV_CODE, DOT_FILE
|
|
|
|
|
|
def charger_instructions_depuis_gitea(nom_fichier="Instructions.md"):
|
|
headers = {"Authorization": f"token {GITEA_TOKEN}"}
|
|
url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/contents/{nom_fichier}?ref={ENV}"
|
|
try:
|
|
response = requests.get(url, headers=headers)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
contenu_md = base64.b64decode(data["content"]).decode("utf-8")
|
|
return contenu_md
|
|
except Exception as e:
|
|
logging.error(f"Erreur chargement instructions Gitea : {e}")
|
|
return None
|
|
|
|
|
|
def recuperer_date_dernier_commit(url):
|
|
headers = {"Authorization": f"token {GITEA_TOKEN}"}
|
|
try:
|
|
response = requests.get(url, headers=headers, timeout=10)
|
|
response.raise_for_status()
|
|
commits = response.json()
|
|
if commits:
|
|
return parser.isoparse(commits[0]["commit"]["author"]["date"])
|
|
except Exception as e:
|
|
logging.error(f"Erreur récupération commit schema : {e}")
|
|
return None
|
|
|
|
|
|
def charger_schema_depuis_gitea(fichier_local="schema_temp.txt"):
|
|
headers = {"Authorization": f"token {GITEA_TOKEN}"}
|
|
url = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_CODE}/contents/{DOT_FILE}?ref={ENV_CODE}"
|
|
try:
|
|
response = requests.get(url, headers=headers, timeout=10)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
remote_last_modified = recuperer_date_dernier_commit(f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_CODE}/commits?path={DOT_FILE}&sha={ENV_CODE}")
|
|
local_last_modified = datetime.fromtimestamp(os.path.getmtime(fichier_local), tz=timezone.utc) if os.path.exists(fichier_local) else None
|
|
|
|
if not local_last_modified or (remote_last_modified and remote_last_modified > local_last_modified):
|
|
dot_text = base64.b64decode(data["content"]).decode("utf-8")
|
|
with open(fichier_local, "w", encoding="utf-8") as f:
|
|
f.write(dot_text)
|
|
|
|
return "OK"
|
|
except Exception as e:
|
|
logging.error(f"Erreur chargement schema Gitea : {e}")
|
|
return None
|
|
|
|
|
|
def charger_arborescence_fiches():
|
|
headers = {"Authorization": f"token {GITEA_TOKEN}"}
|
|
url_base = f"{GITEA_URL}/repos/{ORGANISATION}/{DEPOT_FICHES}/contents/Documents?ref={ENV}"
|
|
|
|
try:
|
|
response = requests.get(url_base, headers=headers)
|
|
response.raise_for_status()
|
|
dossiers = response.json()
|
|
arbo = {}
|
|
|
|
for dossier in sorted(dossiers, key=lambda d: d['name'].lower()):
|
|
if dossier['type'] == 'dir':
|
|
dossier_name = dossier['name']
|
|
url_dossier = dossier['url']
|
|
response_dossier = requests.get(url_dossier, headers=headers)
|
|
response_dossier.raise_for_status()
|
|
fichiers = response_dossier.json()
|
|
fiches = sorted(
|
|
[
|
|
{"nom": f["name"], "download_url": f["download_url"]}
|
|
for f in fichiers if f["name"].endswith(".md")
|
|
],
|
|
key=lambda x: x['nom'].lower()
|
|
)
|
|
arbo[dossier_name] = fiches
|
|
|
|
return arbo
|
|
except Exception as e:
|
|
logging.error(f"Erreur chargement fiches : {e}")
|
|
return {}
|