336 lines
9.4 KiB
Python
336 lines
9.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
from setuphelpers import *
|
|
import json
|
|
import os
|
|
import shutil
|
|
from configparser import ConfigParser
|
|
|
|
import requests
|
|
|
|
|
|
CONFIG_FILE = "ticketing.ini"
|
|
CONFIG_TEMPLATE_FILE = "ticketing.ini.example"
|
|
STATE_FILE = "tickets_state.json"
|
|
DEFAULT_TIMEOUT = 30
|
|
DEFAULT_TICKET_FIELDS = ["id", "name", "stage_id", "partner_id", "user_id", "write_date"]
|
|
DEFAULT_MAX_NOTIFIED_TICKETS = 50
|
|
DEFAULT_TICKET_URL_TEMPLATE = "{odoo_url}/web#id={id}&model={model}&view_type=form"
|
|
|
|
|
|
def get_config_path():
|
|
return makepath(WAPT.private_dir, CONFIG_FILE)
|
|
|
|
|
|
def get_state_path():
|
|
return makepath(WAPT.private_dir, STATE_FILE)
|
|
|
|
|
|
def install():
|
|
config_path = get_config_path()
|
|
|
|
if not isfile(config_path):
|
|
shutil.copyfile(CONFIG_TEMPLATE_FILE, config_path)
|
|
|
|
|
|
def audit():
|
|
conf_wapt = load_config()
|
|
tickets = fetch_odoo_tickets(conf_wapt)
|
|
previous_tickets_state = load_tickets_state(get_state_path())
|
|
tickets_updated = get_updated_tickets(tickets, previous_tickets_state)
|
|
message = build_notification_message(
|
|
tickets_updated,
|
|
previous_tickets_state,
|
|
get_max_notified_tickets(conf_wapt),
|
|
get_ticket_url_template(conf_wapt),
|
|
conf_wapt.get("odoo", "url").rstrip("/"),
|
|
conf_wapt.get("odoo", "model", fallback="helpdesk.ticket"),
|
|
)
|
|
|
|
send_to_rocket(message, conf_wapt=conf_wapt)
|
|
save_tickets_state(get_state_path(), build_tickets_state(tickets))
|
|
return "OK"
|
|
|
|
|
|
def load_config():
|
|
config_path = get_config_path()
|
|
conf_wapt = ConfigParser()
|
|
conf_wapt.read(config_path)
|
|
|
|
required_options = [
|
|
("odoo", "url"),
|
|
("odoo", "database"),
|
|
("odoo", "api_key"),
|
|
("rocket", "webhook_url"),
|
|
]
|
|
if not (
|
|
conf_wapt.has_option("odoo", "uid")
|
|
or conf_wapt.has_option("odoo", "username")
|
|
):
|
|
required_options.append(("odoo", "username"))
|
|
|
|
missing_options = [
|
|
"%s.%s" % (section, option)
|
|
for section, option in required_options
|
|
if not conf_wapt.has_option(section, option)
|
|
]
|
|
|
|
if missing_options:
|
|
raise Exception(
|
|
"Configuration incomplete dans %s : %s"
|
|
% (config_path, ", ".join(missing_options))
|
|
)
|
|
|
|
placeholder_options = [
|
|
"%s.%s" % (section, option)
|
|
for section, option in required_options
|
|
if "CHANGE_ME" in conf_wapt.get(section, option).strip()
|
|
]
|
|
if placeholder_options:
|
|
raise Exception(
|
|
"Configuration a renseigner dans %s : %s"
|
|
% (config_path, ", ".join(placeholder_options))
|
|
)
|
|
|
|
return conf_wapt
|
|
|
|
|
|
def get_http_timeout(conf_wapt):
|
|
return conf_wapt.getint("http", "timeout", fallback=DEFAULT_TIMEOUT)
|
|
|
|
|
|
def get_odoo_fields(conf_wapt):
|
|
fields = [
|
|
field.strip()
|
|
for field in conf_wapt.get("odoo", "fields", fallback=",".join(DEFAULT_TICKET_FIELDS)).split(",")
|
|
if field.strip()
|
|
]
|
|
for required_field in ["id", "write_date"]:
|
|
if required_field not in fields:
|
|
fields.append(required_field)
|
|
return fields
|
|
|
|
|
|
def get_max_notified_tickets(conf_wapt):
|
|
return conf_wapt.getint("notification", "max_tickets", fallback=DEFAULT_MAX_NOTIFIED_TICKETS)
|
|
|
|
|
|
def get_ticket_url_template(conf_wapt):
|
|
return conf_wapt.get("notification", "ticket_url_template", fallback=DEFAULT_TICKET_URL_TEMPLATE)
|
|
|
|
|
|
def build_odoo_payload(service, method, args):
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"method": "call",
|
|
"params": {
|
|
"service": service,
|
|
"method": method,
|
|
"args": args,
|
|
},
|
|
"id": 1,
|
|
}
|
|
|
|
|
|
def call_odoo_jsonrpc(conf_wapt, service, method, args):
|
|
odoo_url = conf_wapt.get("odoo", "url").rstrip("/")
|
|
endpoint = "%s/jsonrpc" % odoo_url
|
|
|
|
response = requests.post(
|
|
endpoint,
|
|
json=build_odoo_payload(service, method, args),
|
|
headers={"Content-Type": "application/json"},
|
|
timeout=get_http_timeout(conf_wapt),
|
|
)
|
|
response.raise_for_status()
|
|
response_data = response.json()
|
|
|
|
if "error" in response_data:
|
|
raise Exception("Erreur Odoo JSON-RPC : %s" % response_data["error"])
|
|
if "result" not in response_data:
|
|
raise Exception("Reponse Odoo invalide : cle 'result' absente")
|
|
|
|
return response_data["result"]
|
|
|
|
|
|
def get_odoo_uid(conf_wapt):
|
|
if conf_wapt.has_option("odoo", "uid"):
|
|
return conf_wapt.getint("odoo", "uid")
|
|
|
|
uid = call_odoo_jsonrpc(
|
|
conf_wapt,
|
|
"common",
|
|
"authenticate",
|
|
[
|
|
conf_wapt.get("odoo", "database"),
|
|
conf_wapt.get("odoo", "username"),
|
|
conf_wapt.get("odoo", "api_key"),
|
|
{},
|
|
],
|
|
)
|
|
if not uid:
|
|
raise Exception("Authentification Odoo impossible")
|
|
return uid
|
|
|
|
|
|
def fetch_odoo_tickets(conf_wapt):
|
|
uid = get_odoo_uid(conf_wapt)
|
|
|
|
return call_odoo_jsonrpc(
|
|
conf_wapt,
|
|
"object",
|
|
"execute_kw",
|
|
[
|
|
conf_wapt.get("odoo", "database"),
|
|
uid,
|
|
conf_wapt.get("odoo", "api_key"),
|
|
conf_wapt.get("odoo", "model", fallback="helpdesk.ticket"),
|
|
"search_read",
|
|
[[]],
|
|
{"fields": get_odoo_fields(conf_wapt)},
|
|
],
|
|
)
|
|
|
|
|
|
def load_tickets_state(state_path):
|
|
if not os.path.exists(state_path):
|
|
return {}
|
|
|
|
with open(state_path, "r") as state_file:
|
|
return json.load(state_file)
|
|
|
|
|
|
def save_tickets_state(state_path, tickets_state):
|
|
with open(state_path, "w") as state_file:
|
|
json.dump(tickets_state, state_file, indent=2, sort_keys=True)
|
|
|
|
|
|
def build_tickets_state(tickets):
|
|
return {
|
|
str(ticket["id"]): ticket["write_date"]
|
|
for ticket in tickets
|
|
}
|
|
|
|
|
|
def get_updated_ticket_ids(tickets, previous_tickets_state):
|
|
return [
|
|
ticket["id"]
|
|
for ticket in get_updated_tickets(tickets, previous_tickets_state)
|
|
]
|
|
|
|
|
|
def get_updated_tickets(tickets, previous_tickets_state):
|
|
tickets_updated = []
|
|
|
|
for ticket in tickets:
|
|
ticket_id = str(ticket["id"])
|
|
ticket_write_date = ticket["write_date"]
|
|
|
|
if previous_tickets_state.get(ticket_id) != ticket_write_date:
|
|
tickets_updated.append(ticket)
|
|
|
|
return tickets_updated
|
|
|
|
|
|
def display_odoo_value(value):
|
|
if value in (False, None, ""):
|
|
return "-"
|
|
if isinstance(value, (list, tuple)):
|
|
if len(value) >= 2:
|
|
return str(value[1])
|
|
if len(value) == 1:
|
|
return str(value[0])
|
|
return "-"
|
|
return str(value)
|
|
|
|
|
|
def build_ticket_url(ticket, ticket_url_template, odoo_url, model):
|
|
return ticket_url_template.format(
|
|
id=ticket["id"],
|
|
odoo_url=odoo_url,
|
|
model=model,
|
|
)
|
|
|
|
|
|
def format_ticket_line(ticket, ticket_url_template, odoo_url, model):
|
|
assignee = display_odoo_value(ticket.get("user_id"))
|
|
if assignee == "-":
|
|
assignee = "Personne"
|
|
|
|
return "%s : Sujet : %s gere par %s : [cliquer ici pour ouvrir le ticket](%s)" % (
|
|
ticket["id"],
|
|
display_odoo_value(ticket.get("name", "Sans titre")),
|
|
assignee,
|
|
build_ticket_url(ticket, ticket_url_template, odoo_url, model),
|
|
)
|
|
|
|
|
|
def format_detailed_ticket_line(ticket, previous_tickets_state):
|
|
ticket_id = str(ticket["id"])
|
|
previous_write_date = previous_tickets_state.get(ticket_id)
|
|
change_type = "nouveau" if previous_write_date is None else "modifie"
|
|
parts = [
|
|
"#%s" % ticket["id"],
|
|
display_odoo_value(ticket.get("name", "Sans titre")),
|
|
"type: %s" % change_type,
|
|
"statut: %s" % display_odoo_value(ticket.get("stage_id")),
|
|
"client: %s" % display_odoo_value(ticket.get("partner_id")),
|
|
"assigne: %s" % display_odoo_value(ticket.get("user_id")),
|
|
"date: %s" % display_odoo_value(ticket.get("write_date")),
|
|
]
|
|
if previous_write_date is not None:
|
|
parts.append("ancienne date: %s" % previous_write_date)
|
|
return "- " + " | ".join(parts)
|
|
|
|
|
|
def build_notification_message(tickets_updated, previous_tickets_state, max_tickets, ticket_url_template, odoo_url, model):
|
|
if not tickets_updated:
|
|
return "Aucun ticket n'a ete mis a jour depuis la derniere verification."
|
|
|
|
displayed_tickets = tickets_updated[:max_tickets]
|
|
hidden_count = len(tickets_updated) - len(displayed_tickets)
|
|
lines = [
|
|
"Les tickets suivants ont ete mis a jour depuis la derniere verification :",
|
|
]
|
|
lines.extend([
|
|
format_ticket_line(ticket, ticket_url_template, odoo_url, model)
|
|
for ticket in displayed_tickets
|
|
])
|
|
|
|
if hidden_count > 0:
|
|
lines.extend([
|
|
"",
|
|
"... %s ticket(s) supplementaire(s) non affiche(s) sur %s au total." % (
|
|
hidden_count,
|
|
len(tickets_updated),
|
|
),
|
|
])
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def send_to_rocket(message_text, attachments=None, conf_wapt=None):
|
|
"""
|
|
Envoie un message a Rocket.Chat via un webhook.
|
|
|
|
:param message_text: Texte du message a envoyer
|
|
:param attachments: Liste de pieces jointes (facultatif)
|
|
:param conf_wapt: Configuration deja chargee (facultatif)
|
|
"""
|
|
if conf_wapt is None:
|
|
conf_wapt = load_config()
|
|
|
|
message = {
|
|
"text": message_text,
|
|
}
|
|
if attachments:
|
|
message["attachments"] = attachments
|
|
|
|
response = requests.post(
|
|
conf_wapt.get("rocket", "webhook_url"),
|
|
json=message,
|
|
headers={"Content-Type": "application/json"},
|
|
timeout=get_http_timeout(conf_wapt),
|
|
)
|
|
response.raise_for_status()
|
|
print("Message envoye avec succes.")
|