1162 lines
48 KiB
Python
Executable File
1162 lines
48 KiB
Python
Executable File
# -*- coding: utf-8 -*-
|
|
|
|
import json
|
|
import mimetypes
|
|
import posixpath
|
|
import tempfile
|
|
import base64
|
|
import logging
|
|
from dateutil.relativedelta import relativedelta
|
|
from urllib.error import URLError
|
|
from urllib.parse import quote, unquote, urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
from odoo import http
|
|
from odoo.http import request
|
|
from webdav4.fsspec import WebdavFileSystem
|
|
from werkzeug.exceptions import NotFound
|
|
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
_logger.setLevel(logging.DEBUG)
|
|
_logger.setloging(True)
|
|
|
|
|
|
|
|
class AdPortalController(http.Controller):
|
|
|
|
MEDIA_ROOT_SUFFIX = '02_Fertig'
|
|
|
|
def _json_response(self, payload, status=200):
|
|
return request.make_response(
|
|
json.dumps(payload),
|
|
headers=[('Content-Type', 'application/json')],
|
|
status=status,
|
|
)
|
|
|
|
def _get_adportal_settings(self):
|
|
return request.env['dss.settings'].sudo().search([], limit=1)
|
|
|
|
def _get_visible_graphic_templates(self):
|
|
return request.env['dss.mediatypes'].sudo().search([
|
|
('adportal_visible', '=', True),
|
|
], order='medianame asc, id asc')
|
|
|
|
def _resolve_graphic_resolution(self, template):
|
|
default_resolution = {
|
|
'width': 1024,
|
|
'height': 1024,
|
|
'label': '1024 x 1024',
|
|
}
|
|
if not template:
|
|
return default_resolution
|
|
|
|
width = int(template.maxsize_w or 0)
|
|
height = int(template.maxsize_h or 0)
|
|
if width <= 0 or height <= 0:
|
|
return default_resolution
|
|
|
|
return {
|
|
'width': width,
|
|
'height': height,
|
|
'label': '%s x %s' % (width, height),
|
|
}
|
|
|
|
def _get_openai_image_size(self, model, width, height):
|
|
width = int(width or 0)
|
|
height = int(height or 0)
|
|
if width <= 0 or height <= 0:
|
|
return '1024x1024'
|
|
|
|
portrait = height > width
|
|
square = width == height
|
|
model_name = (model or '').strip()
|
|
|
|
if square:
|
|
return '1024x1024'
|
|
if portrait:
|
|
if model_name == 'dall-e-3':
|
|
return '1024x1792'
|
|
return '1024x1536'
|
|
if model_name == 'dall-e-3':
|
|
return '1792x1024'
|
|
return '1536x1024'
|
|
|
|
def _build_graphic_prompt(
|
|
self,
|
|
settings,
|
|
company_name,
|
|
company_address='',
|
|
instruction='',
|
|
resolution_label='',
|
|
mediatype_name='',
|
|
mediatype_prompt='',
|
|
company_context='',
|
|
):
|
|
base_prompt = (settings.adportal_graphics_base_prompt or '').strip() if settings else ''
|
|
parts = []
|
|
if base_prompt:
|
|
parts.append(base_prompt)
|
|
if mediatype_name:
|
|
parts.append('Medientyp: %s' % mediatype_name)
|
|
if mediatype_prompt:
|
|
parts.append('Medientyp KI Prompt: %s' % mediatype_prompt)
|
|
parts.append('Firma: %s' % company_name)
|
|
if company_address:
|
|
parts.append('Adresse: %s' % company_address)
|
|
if company_context:
|
|
parts.append(company_context)
|
|
if resolution_label:
|
|
parts.append('Aufloesung: %s Pixel' % resolution_label)
|
|
if instruction:
|
|
parts.append('Aenderung: %s' % instruction)
|
|
return '\n\n'.join(parts)
|
|
|
|
def _build_graphic_prompt_with_chatgpt(
|
|
self,
|
|
settings,
|
|
company_name,
|
|
company_address='',
|
|
instruction='',
|
|
resolution_label='',
|
|
mediatype_name='',
|
|
mediatype_prompt='',
|
|
company_context='',
|
|
):
|
|
prompt = self._build_graphic_prompt(
|
|
settings,
|
|
company_name,
|
|
company_address=company_address,
|
|
instruction=instruction,
|
|
resolution_label=resolution_label,
|
|
mediatype_name=mediatype_name,
|
|
mediatype_prompt=mediatype_prompt,
|
|
company_context=company_context,
|
|
)
|
|
return prompt, ''
|
|
|
|
def _get_company_context_from_google(self, settings, company_name):
|
|
if not settings:
|
|
return '', ''
|
|
|
|
api_key = (settings.adportal_google_api_key or '').strip()
|
|
cse_id = (settings.adportal_google_cse_id or '').strip()
|
|
if not api_key or not cse_id:
|
|
return '', ''
|
|
|
|
query_url = 'https://www.googleapis.com/customsearch/v1?%s' % urlencode({
|
|
'key': api_key,
|
|
'cx': cse_id,
|
|
'q': company_name,
|
|
'num': 3,
|
|
})
|
|
req = Request(query_url, headers={'Accept': 'application/json'})
|
|
try:
|
|
with urlopen(req, timeout=10) as response:
|
|
payload = json.loads(response.read().decode('utf-8'))
|
|
except Exception:
|
|
return '', 'Google Firmeninformationen konnten nicht geladen werden.'
|
|
|
|
items = payload.get('items') or []
|
|
if not items:
|
|
return '', ''
|
|
|
|
snippets = []
|
|
for item in items[:3]:
|
|
title = (item.get('title') or '').strip()
|
|
snippet = (item.get('snippet') or '').strip().replace('\n', ' ')
|
|
link = (item.get('link') or '').strip()
|
|
info_parts = [part for part in (title, snippet, link) if part]
|
|
if info_parts:
|
|
snippets.append(' - '.join(info_parts))
|
|
|
|
if not snippets:
|
|
return '', ''
|
|
|
|
return 'Google Firmenkontext:\n%s' % '\n'.join(snippets), ''
|
|
|
|
def _create_graphic_with_chatgpt(self, settings, prompt, current_image='', resolution_width=0, resolution_height=0):
|
|
if not settings:
|
|
return None, 'Keine AD Portal Settings gefunden.'
|
|
|
|
try:
|
|
from openai import OpenAI
|
|
except Exception:
|
|
return None, 'openai package not available'
|
|
|
|
settings_icp = request.env['ir.config_parameter'].sudo()
|
|
api_key = (settings.adportal_graphics_api_key or '').strip() or (settings_icp.get_param('digitalsignage_ai.api_key') or '').strip()
|
|
if not api_key:
|
|
return None, 'ChatGPT API Key fehlt.'
|
|
|
|
base_url = (settings_icp.get_param('digitalsignage_ai.base_url') or 'https://api.openai.com/v1').rstrip('/')
|
|
model = (settings.adportal_graphics_chatgpt_model or 'gpt-image-1').strip()
|
|
image_size = self._get_openai_image_size(model, resolution_width, resolution_height)
|
|
|
|
try:
|
|
client = OpenAI(api_key=api_key, base_url=base_url)
|
|
if current_image:
|
|
image_bytes = None
|
|
if current_image.startswith('data:image/') and ';base64,' in current_image:
|
|
image_bytes = base64.b64decode(current_image.split(';base64,', 1)[1])
|
|
elif current_image.startswith('data:') and ';base64,' in current_image:
|
|
image_bytes = base64.b64decode(current_image.split(';base64,', 1)[1])
|
|
elif current_image.startswith('http://') or current_image.startswith('https://'):
|
|
with urlopen(Request(current_image, headers={'Accept': 'application/octet-stream'}), timeout=30) as response:
|
|
image_bytes = response.read()
|
|
|
|
if image_bytes:
|
|
with tempfile.NamedTemporaryFile(suffix='.png') as image_file:
|
|
image_file.write(image_bytes)
|
|
image_file.flush()
|
|
with open(image_file.name, 'rb') as stream:
|
|
response = client.images.edits(
|
|
model=model,
|
|
image=stream,
|
|
prompt=prompt,
|
|
size=image_size,
|
|
)
|
|
else:
|
|
response = client.images.generate(
|
|
model=model,
|
|
prompt=prompt,
|
|
size=image_size,
|
|
)
|
|
else:
|
|
response = client.images.generate(
|
|
model=model,
|
|
prompt=prompt,
|
|
size=image_size,
|
|
)
|
|
|
|
if response and getattr(response, 'data', None):
|
|
first_item = response.data[0]
|
|
image_url = getattr(first_item, 'url', '') or ''
|
|
image_b64 = getattr(first_item, 'b64_json', '') or ''
|
|
if image_url:
|
|
return {'image_url': image_url, 'image_data': ''}, ''
|
|
if image_b64:
|
|
return {'image_url': '', 'image_data': 'data:image/png;base64,%s' % image_b64}, ''
|
|
return None, 'OpenAI hat kein Bild geliefert.'
|
|
except Exception as exc:
|
|
return None, str(exc)
|
|
|
|
def _validate_company_with_google(self, settings, company_name):
|
|
if not settings or not settings.adportal_google_company_check:
|
|
return True, '', [{
|
|
'id': 'manual',
|
|
'name': company_name,
|
|
'street': '',
|
|
'city': '',
|
|
'postal_code': '',
|
|
'region': '',
|
|
'country': '',
|
|
'full_address': company_name,
|
|
}]
|
|
|
|
api_key = (settings.adportal_google_api_key or '').strip()
|
|
cse_id = (settings.adportal_google_cse_id or '').strip()
|
|
if not api_key or not cse_id:
|
|
return False, 'Google Pruefung aktiv, aber API Key oder CSE ID fehlt in den Settings.', []
|
|
|
|
_logger.info('ADPORTAL : Validating company "%s" with Google Search API', company_name)
|
|
|
|
|
|
import requests
|
|
|
|
url = "https://local-business-data.p.rapidapi.com/search"
|
|
|
|
querystring = {"query":"Firma wie "+company_name,"limit":"20","zoom":"13","language":"en","region":"de","extract_emails_and_contacts":"false"}
|
|
|
|
headers = {
|
|
"x-rapidapi-key": "6eb64b4d3bmshe5bc9fc666cce68p13a32djsna3f717099ebf",
|
|
"x-rapidapi-host": "local-business-data.p.rapidapi.com",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
response = requests.get(url, headers=headers, params=querystring)
|
|
data = response.json()
|
|
|
|
_logger.info('ADPORTAL : Google Search API response: %s', str(data))
|
|
|
|
status_value = str(data.get('status') or '').strip().upper()
|
|
if status_value != 'OK':
|
|
return False, 'Google Firmenpruefung nicht OK (status: %s).' % (status_value or 'leer'), []
|
|
|
|
records = data.get('data') or []
|
|
if not isinstance(records, list) or not records:
|
|
return False, 'Keine passenden Firmenadressen gefunden.', []
|
|
|
|
addresses = []
|
|
for idx, row in enumerate(records):
|
|
if not isinstance(row, dict):
|
|
continue
|
|
|
|
name = (row.get('name') or row.get('business_name') or company_name or '').strip()
|
|
street = (row.get('street') or row.get('address') or '').strip()
|
|
city = (row.get('city') or '').strip()
|
|
postal_code = (row.get('zip') or row.get('postal_code') or row.get('postcode') or '').strip()
|
|
region = (row.get('state') or row.get('region') or '').strip()
|
|
country = (row.get('country') or '').strip()
|
|
|
|
parts = [part for part in (street, postal_code, city, region, country) if part]
|
|
full_address = ', '.join(parts)
|
|
if not full_address:
|
|
full_address = (row.get('formatted_address') or '').strip()
|
|
|
|
addresses.append({
|
|
'id': str(row.get('place_id') or row.get('cid') or (idx + 1)),
|
|
'name': name or company_name,
|
|
'street': street,
|
|
'city': city,
|
|
'postal_code': postal_code,
|
|
'region': region,
|
|
'country': country,
|
|
'full_address': full_address,
|
|
})
|
|
|
|
if not addresses:
|
|
return False, 'Keine auswertbaren Firmenadressen gefunden.', []
|
|
return True, '', addresses
|
|
|
|
|
|
|
|
def _validate_company_address(self, settings, company_name, company_address):
|
|
address_value = (company_address or '').strip()
|
|
if not address_value:
|
|
return False, 'Bitte eine Adresse eingeben.'
|
|
if len(address_value) < 6:
|
|
return False, 'Adresse ist zu kurz. Bitte vollstaendige Adresse eingeben.'
|
|
|
|
if not settings or not settings.adportal_google_company_check:
|
|
return True, ''
|
|
|
|
api_key = (settings.adportal_google_api_key or '').strip()
|
|
cse_id = (settings.adportal_google_cse_id or '').strip()
|
|
if not api_key or not cse_id:
|
|
return False, 'Google Pruefung aktiv, aber API Key oder CSE ID fehlt in den Settings.'
|
|
|
|
query_text = '%s %s' % ((company_name or '').strip(), address_value)
|
|
query_url = 'https://www.googleapis.com/customsearch/v1?%s' % urlencode({
|
|
'key': api_key,
|
|
'cx': cse_id,
|
|
'q': query_text,
|
|
'num': 1,
|
|
})
|
|
req = Request(query_url, headers={'Accept': 'application/json'})
|
|
try:
|
|
with urlopen(req, timeout=10) as response:
|
|
payload = json.loads(response.read().decode('utf-8'))
|
|
except Exception:
|
|
return False, 'Adresspruefung konnte nicht ausgefuehrt werden.'
|
|
|
|
if payload.get('items'):
|
|
return True, ''
|
|
return False, 'Adresse konnte nicht bestaetigt werden. Bitte Adresse pruefen.'
|
|
|
|
def _call_graphics_engine(self, settings, payload):
|
|
if not settings:
|
|
return None, 'Keine AD Portal Settings gefunden.'
|
|
|
|
if settings.adportal_graphics_use_chatgpt:
|
|
image_result, error = self._create_graphic_with_chatgpt(
|
|
settings,
|
|
payload.get('prompt') or '',
|
|
current_image=payload.get('current_image') or '',
|
|
resolution_width=payload.get('resolution_width') or 0,
|
|
resolution_height=payload.get('resolution_height') or 0,
|
|
)
|
|
if error:
|
|
return None, error
|
|
return image_result, ''
|
|
|
|
api_url = (settings.adportal_graphics_api_url or '').strip()
|
|
api_key = (settings.adportal_graphics_api_key or '').strip()
|
|
timeout = int(settings.adportal_graphics_api_timeout or 90)
|
|
|
|
if not api_url:
|
|
return None, 'Grafik API URL fehlt in den AD Portal Settings.'
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
}
|
|
if api_key:
|
|
headers['Authorization'] = 'Bearer %s' % api_key
|
|
headers['X-API-Key'] = api_key
|
|
|
|
req = Request(api_url, data=json.dumps(payload).encode('utf-8'), headers=headers, method='POST')
|
|
try:
|
|
with urlopen(req, timeout=max(5, timeout)) as response:
|
|
body = response.read().decode('utf-8')
|
|
return json.loads(body or '{}'), ''
|
|
except URLError as exc:
|
|
return None, 'Grafik API nicht erreichbar: %s' % str(exc)
|
|
except Exception as exc:
|
|
return None, 'Grafik API Fehler: %s' % str(exc)
|
|
|
|
def _extract_image_from_ai_result(self, data):
|
|
if not data:
|
|
return '', ''
|
|
|
|
if isinstance(data, dict):
|
|
image_data = data.get('image_data')
|
|
if isinstance(image_data, str) and image_data.strip():
|
|
return '', image_data.strip()
|
|
|
|
direct_url_keys = ('image_url', 'url', 'output_image_url')
|
|
for key in direct_url_keys:
|
|
value = data.get(key) if isinstance(data, dict) else None
|
|
if isinstance(value, str) and value.strip():
|
|
return value.strip(), ''
|
|
|
|
if isinstance(data, dict):
|
|
image_block = data.get('image')
|
|
if isinstance(image_block, dict):
|
|
image_url = image_block.get('url')
|
|
if isinstance(image_url, str) and image_url.strip():
|
|
return image_url.strip(), ''
|
|
b64 = image_block.get('b64_json') or image_block.get('base64')
|
|
if isinstance(b64, str) and b64.strip():
|
|
return '', 'data:image/png;base64,%s' % b64.strip()
|
|
|
|
images = data.get('images')
|
|
if isinstance(images, list) and images:
|
|
first = images[0]
|
|
if isinstance(first, str) and first.strip():
|
|
return first.strip(), ''
|
|
if isinstance(first, dict):
|
|
image_url = first.get('url')
|
|
if isinstance(image_url, str) and image_url.strip():
|
|
return image_url.strip(), ''
|
|
b64 = first.get('b64_json') or first.get('base64')
|
|
if isinstance(b64, str) and b64.strip():
|
|
return '', 'data:image/png;base64,%s' % b64.strip()
|
|
|
|
b64_top = data.get('image_base64') or data.get('b64_json')
|
|
if isinstance(b64_top, str) and b64_top.strip():
|
|
return '', 'data:image/png;base64,%s' % b64_top.strip()
|
|
|
|
return '', ''
|
|
|
|
def _user_has_portal_right(self, user, field_name):
|
|
if field_name not in (getattr(user, '_fields', {}) or {}):
|
|
return True
|
|
return bool(getattr(user, field_name, True))
|
|
|
|
def _ensure_portal_right(self, user, field_name):
|
|
if not self._user_has_portal_right(user, field_name):
|
|
raise NotFound()
|
|
|
|
def _get_portal_right_label(self, user, field_name):
|
|
labels = {
|
|
'portal_right_contract_overview': 'Vertragsanzeige',
|
|
'portal_right_contract_detail': 'Vertragsdetails',
|
|
'portal_right_media_view': 'Medienanzeige',
|
|
'portal_right_ranking_view': 'Rankinganzeige',
|
|
'portal_right_graphic_examples': 'Grafikbeispiele anzeigen/erstellen',
|
|
}
|
|
return labels.get(field_name, field_name)
|
|
|
|
def _normalize_cloud_path(self, path):
|
|
path_value = (path or '').strip().replace('\\', '/')
|
|
if not path_value:
|
|
return ''
|
|
normalized = posixpath.normpath('/' + path_value).lstrip('/')
|
|
return '' if normalized == '.' else normalized
|
|
|
|
def _join_cloud_paths(self, left, right):
|
|
if not left:
|
|
return self._normalize_cloud_path(right)
|
|
if not right:
|
|
return self._normalize_cloud_path(left)
|
|
return self._normalize_cloud_path('%s/%s' % (left.rstrip('/'), right.lstrip('/')))
|
|
|
|
def _ensure_contract_portal_access(self, contract, contract_model, user):
|
|
if not contract:
|
|
raise NotFound()
|
|
|
|
contract_fields = getattr(contract_model, '_fields', {}) or {}
|
|
is_creator = True
|
|
if 'contract_creator' in contract_fields:
|
|
if contract.contract_creator:
|
|
is_creator = contract.contract_creator.id == user.id
|
|
|
|
is_writer_with_project_partner = False
|
|
partner_id = user.partner_id.id if getattr(user, 'partner_id', False) else False
|
|
if partner_id and 'contract_writer' in contract_fields and contract.contract_writer and contract.contract_writer.id == partner_id:
|
|
project = contract.project
|
|
project_fields = getattr(project, '_fields', {}) if project else {}
|
|
if project and 'vertriebspartner' in (project_fields or {}):
|
|
is_writer_with_project_partner = partner_id in project.vertriebspartner.ids
|
|
|
|
if not (is_creator or is_writer_with_project_partner):
|
|
raise NotFound()
|
|
|
|
def _get_user_contract_visibility_domain(self, user, require_project=False, project_id=False):
|
|
base_domain = []
|
|
if require_project:
|
|
base_domain.append(('project', '!=', False))
|
|
if project_id:
|
|
base_domain.append(('project', '=', project_id))
|
|
|
|
creator_domain = [('contract_creator', '=', user.id)]
|
|
partner_id = user.partner_id.id if getattr(user, 'partner_id', False) else False
|
|
if not partner_id:
|
|
return base_domain + creator_domain
|
|
|
|
# Additional visibility: user is contract_writer and is in project.vertriebspartner.
|
|
writer_and_partner_domain = ['&',
|
|
('contract_writer', '=', partner_id),
|
|
('project.vertriebspartner', 'in', [partner_id]),
|
|
]
|
|
return base_domain + ['|'] + creator_domain + writer_and_partner_domain
|
|
|
|
def _compute_contract_end_without_extension(self, contract):
|
|
_logger.warning('ADPORTAL: compute end (no extension) CALLED for contract id=%s', contract.id if contract else 'n/a')
|
|
if not contract:
|
|
return False
|
|
|
|
start_date = contract.start_date or contract.contract_date
|
|
if not start_date:
|
|
return False
|
|
|
|
if contract.contract_iscanceled and contract.ads_last_todo_state_until:
|
|
_logger.warning('ADPORTAL: contract id=%s is canceled, using ads_last_todo_state_until=%s', contract.id, contract.ads_last_todo_state_until)
|
|
return contract.ads_last_todo_state_until
|
|
|
|
runtime_months = int(contract.runtime_m or 0) + int(contract.runtime_bonus_m or 0)
|
|
if contract.contract_auto_extend:
|
|
try:
|
|
extension_months = int((contract.contract_auto_extend_time or '0').strip() or 0)
|
|
except Exception:
|
|
extension_months = 0
|
|
runtime_months += extension_months
|
|
_logger.warning('ADPORTAL: contract id=%s has auto extension=%s months', contract.id, extension_months)
|
|
|
|
_logger.warning('ADPORTAL: compute end inputs contract id=%s start_date=%s contract_date=%s runtime_m=%s runtime_bonus_m=%s total_months=%s', contract.id, contract.start_date, contract.contract_date, contract.runtime_m, contract.runtime_bonus_m, runtime_months)
|
|
return start_date + relativedelta(months=runtime_months)
|
|
|
|
def _get_webdav_connection_data(self):
|
|
settings = request.env['dss.settings'].sudo().search([], limit=1)
|
|
cloud_base = (settings.def_cloud_url_base or '').strip() if settings else ''
|
|
if not cloud_base:
|
|
return None, None, None
|
|
|
|
dav_base = cloud_base
|
|
marker = 'index.php/apps/files/?dir='
|
|
if marker in dav_base:
|
|
dav_base = '%sremote.php/dav/files/OdooDav/' % dav_base.split(marker)[0]
|
|
elif '/remote.php/dav/files/' not in dav_base:
|
|
dav_base = dav_base.rstrip('/') + '/remote.php/dav/files/OdooDav/'
|
|
|
|
config = request.env['ir.config_parameter'].sudo()
|
|
webdav_user = config.get_param('digitalsignage.webdav_user') or 'odooClient@logumedia.de'
|
|
webdav_password = config.get_param('digitalsignage.webdav_password') or 'lm2020#OdooDav'
|
|
return dav_base, webdav_user, webdav_password
|
|
|
|
def _get_contract_cloud_root(self, contract):
|
|
cloud_root = contract.cloudlink or ''
|
|
if not cloud_root:
|
|
return ''
|
|
|
|
settings = request.env['dss.settings'].sudo().search([], limit=1)
|
|
if settings:
|
|
try:
|
|
cloud_root = settings._get_path_converted(cloud_root, contract) or cloud_root
|
|
except Exception:
|
|
cloud_root = contract.cloudlink or ''
|
|
|
|
normalized = self._normalize_cloud_path(cloud_root)
|
|
if not normalized:
|
|
return ''
|
|
if normalized.split('/')[-1] == self.MEDIA_ROOT_SUFFIX:
|
|
return normalized
|
|
return self._join_cloud_paths(normalized, self.MEDIA_ROOT_SUFFIX)
|
|
|
|
def _build_media_breadcrumbs(self, contract_uuid, current_rel_path):
|
|
breadcrumbs = [
|
|
{
|
|
'name': self.MEDIA_ROOT_SUFFIX,
|
|
'path': '',
|
|
'href': '/adportal/contracts/%s' % contract_uuid,
|
|
'is_current': current_rel_path == '',
|
|
}
|
|
]
|
|
if not current_rel_path:
|
|
return breadcrumbs
|
|
|
|
segments = [segment for segment in current_rel_path.split('/') if segment]
|
|
running_path = ''
|
|
for index, segment in enumerate(segments):
|
|
running_path = self._join_cloud_paths(running_path, segment)
|
|
breadcrumbs.append({
|
|
'name': segment,
|
|
'path': running_path,
|
|
'href': '/adportal/contracts/%s?media_path=%s' % (contract_uuid, quote(running_path)),
|
|
'is_current': index == len(segments) - 1,
|
|
})
|
|
return breadcrumbs
|
|
|
|
def _get_contract_media_entries(self, contract, relative_path=''):
|
|
dav_base, webdav_user, webdav_password = self._get_webdav_connection_data()
|
|
cloud_root = self._get_contract_cloud_root(contract)
|
|
|
|
if not dav_base or not cloud_root:
|
|
return {
|
|
'enabled': False,
|
|
'entries': [],
|
|
'current_path': '',
|
|
'parent_path': '',
|
|
'error': '',
|
|
}
|
|
|
|
current_rel_path = self._normalize_cloud_path(unquote(relative_path or ''))
|
|
target_path = self._join_cloud_paths(cloud_root, current_rel_path)
|
|
|
|
try:
|
|
client = WebdavFileSystem(dav_base, auth=(webdav_user, webdav_password))
|
|
raw_entries = client.ls(target_path, detail=True)
|
|
except Exception as exc:
|
|
return {
|
|
'enabled': False,
|
|
'entries': [],
|
|
'current_path': current_rel_path,
|
|
'parent_path': '',
|
|
'error': str(exc),
|
|
}
|
|
|
|
entries = []
|
|
for item in raw_entries:
|
|
if isinstance(item, dict):
|
|
full_name = self._normalize_cloud_path(item.get('name') or item.get('path') or '')
|
|
item_type = str(item.get('type') or '')
|
|
size = item.get('size')
|
|
else:
|
|
full_name = self._normalize_cloud_path(str(item))
|
|
item_type = ''
|
|
size = None
|
|
|
|
if not full_name or full_name == target_path:
|
|
continue
|
|
|
|
if full_name.startswith(cloud_root):
|
|
relative_from_root = full_name[len(cloud_root):].lstrip('/')
|
|
else:
|
|
relative_from_root = full_name
|
|
|
|
name = full_name.rstrip('/').split('/')[-1]
|
|
if not name:
|
|
continue
|
|
|
|
is_dir = item_type.lower() in ('directory', 'dir') or full_name.endswith('/')
|
|
encoded_path = quote(relative_from_root)
|
|
nav_href = '/adportal/contracts/%s?media_path=%s' % (contract.uuid, encoded_path)
|
|
href = nav_href if is_dir else '/adportal/contracts/%s/media?path=%s' % (contract.uuid, encoded_path)
|
|
|
|
entries.append({
|
|
'name': name,
|
|
'is_dir': is_dir,
|
|
'size': size,
|
|
'href': href,
|
|
'nav_href': nav_href,
|
|
'nav_path': relative_from_root,
|
|
})
|
|
|
|
entries = sorted(entries, key=lambda entry: (not entry['is_dir'], entry['name'].lower()))
|
|
|
|
parent_path = ''
|
|
if current_rel_path:
|
|
parent_path = posixpath.dirname(current_rel_path)
|
|
if parent_path == '.':
|
|
parent_path = ''
|
|
|
|
return {
|
|
'enabled': True,
|
|
'entries': entries,
|
|
'current_path': current_rel_path,
|
|
'parent_path': parent_path,
|
|
'parent_href': '/adportal/contracts/%s' % contract.uuid if not parent_path else '/adportal/contracts/%s?media_path=%s' % (contract.uuid, quote(parent_path)),
|
|
'panel_url': '/adportal/contracts/%s/media/panel' % contract.uuid,
|
|
'breadcrumbs': self._build_media_breadcrumbs(contract.uuid, current_rel_path),
|
|
'error': '',
|
|
}
|
|
|
|
def _build_media_values(self, contract, media_path=''):
|
|
try:
|
|
media_info = self._get_contract_media_entries(contract, media_path)
|
|
except Exception:
|
|
media_info = {
|
|
'enabled': False,
|
|
'entries': [],
|
|
'current_path': '',
|
|
'parent_path': '',
|
|
'parent_href': '/adportal/contracts/%s' % contract.uuid,
|
|
'panel_url': '/adportal/contracts/%s/media/panel' % contract.uuid,
|
|
'breadcrumbs': self._build_media_breadcrumbs(contract.uuid, ''),
|
|
'error': '',
|
|
}
|
|
|
|
return {
|
|
'contract': contract,
|
|
'media_enabled': media_info['enabled'],
|
|
'media_entries': media_info['entries'],
|
|
'media_current_path': media_info['current_path'],
|
|
'media_parent_path': media_info['parent_path'],
|
|
'media_parent_href': media_info.get('parent_href', '/adportal/contracts/%s' % contract.uuid),
|
|
'media_panel_url': media_info.get('panel_url', '/adportal/contracts/%s/media/panel' % contract.uuid),
|
|
'media_breadcrumbs': media_info.get('breadcrumbs', self._build_media_breadcrumbs(contract.uuid, '')),
|
|
'media_error': media_info['error'],
|
|
}
|
|
|
|
@http.route(['/my'], type='http', auth='user', website=True, sitemap=False)
|
|
def adportal_my(self, **kwargs):
|
|
return request.redirect('/adportal/dashboard')
|
|
|
|
@http.route(['/adportal/dashboard'], type='http', auth='user', website=True)
|
|
def adportal_dashboard(self, **kwargs):
|
|
user = request.env.user
|
|
contract_model = request.env['dss.contracts'].sudo()
|
|
can_show_contracts = self._user_has_portal_right(user, 'portal_right_contract_overview')
|
|
can_show_graphic_examples = self._user_has_portal_right(user, 'portal_right_graphic_examples')
|
|
|
|
draft_domain = [
|
|
('contract_creator', '=', user.id),
|
|
('web_contract_input_notfinished', '=', True),
|
|
]
|
|
recent_domain = self._get_user_contract_visibility_domain(user)
|
|
|
|
draft_count = 0
|
|
total_count = 0
|
|
recent_contracts = contract_model.browse([])
|
|
if can_show_contracts:
|
|
draft_count = contract_model.search_count(draft_domain)
|
|
total_count = contract_model.search_count(recent_domain)
|
|
recent_contracts = contract_model.search(recent_domain, limit=8, order='write_date desc, id desc')
|
|
|
|
settings = request.env['dss.settings'].sudo().search([], limit=1)
|
|
show_foreign_ranking_names = True if not settings else bool(settings.adportal_ranking_show_foreign_names)
|
|
|
|
ranking_rows = []
|
|
if self._user_has_portal_right(user, 'portal_right_ranking_view'):
|
|
ranking_profiles = request.env['dss.portal.user.profile'].sudo().search([
|
|
('ranking_enabled', '=', True),
|
|
('user_id.active', '=', True),
|
|
], order='ranking_sum desc, ranking_position asc, id asc')
|
|
|
|
max_sum = 0.0
|
|
for ranking_profile in ranking_profiles:
|
|
current_sum = float(ranking_profile.ranking_sum or 0.0)
|
|
if current_sum > max_sum:
|
|
max_sum = current_sum
|
|
|
|
for ranking_profile in ranking_profiles:
|
|
ranking_user = ranking_profile.user_id
|
|
current_sum = float(ranking_profile.ranking_sum or 0.0)
|
|
if max_sum > 0 and current_sum > 0:
|
|
stars = int(round((current_sum / max_sum) * 8.0))
|
|
stars = max(1, stars)
|
|
else:
|
|
stars = 0
|
|
stars = max(0, min(8, stars))
|
|
|
|
display_name = ranking_user.name
|
|
if (not show_foreign_ranking_names) and ranking_user.id != user.id:
|
|
display_name = '*****'
|
|
|
|
ranking_rows.append({
|
|
'name': display_name,
|
|
'stars': stars,
|
|
'sum': current_sum,
|
|
})
|
|
|
|
values = {
|
|
'username': user.name,
|
|
'draft_count': draft_count,
|
|
'total_count': total_count,
|
|
'recent_contracts': recent_contracts,
|
|
'show_contract_overview': can_show_contracts,
|
|
'show_graphic_examples': can_show_graphic_examples,
|
|
'show_ranking': self._user_has_portal_right(user, 'portal_right_ranking_view'),
|
|
'ranking_rows': ranking_rows,
|
|
'page_name': 'adportal_dashboard',
|
|
}
|
|
return request.render('DigitalSignage.adportal_dashboard', values)
|
|
|
|
@http.route(['/adportal/graphic-examples'], type='http', auth='user', website=True)
|
|
def adportal_graphic_examples(self, **kwargs):
|
|
user = request.env.user
|
|
self._ensure_portal_right(user, 'portal_right_graphic_examples')
|
|
|
|
graphic_templates = self._get_visible_graphic_templates()
|
|
suggestions = request.env['dss.contracts'].sudo().search_read(
|
|
[('client_short_company', '!=', False)],
|
|
['client_short_company'],
|
|
limit=200,
|
|
order='write_date desc, id desc',
|
|
)
|
|
company_suggestions = []
|
|
seen = set()
|
|
for row in suggestions:
|
|
company_name = (row.get('client_short_company') or '').strip()
|
|
if not company_name:
|
|
continue
|
|
lowered = company_name.lower()
|
|
if lowered in seen:
|
|
continue
|
|
seen.add(lowered)
|
|
company_suggestions.append(company_name)
|
|
|
|
values = {
|
|
'username': user.name,
|
|
'graphic_templates': graphic_templates,
|
|
'company_suggestions': company_suggestions,
|
|
'page_name': 'adportal_graphic_examples',
|
|
}
|
|
return request.render('DigitalSignage.adportal_graphic_examples', values)
|
|
|
|
@http.route(['/adportal/graphic-examples/start'], type='http', auth='user', methods=['POST'], csrf=False)
|
|
def adportal_graphic_examples_start(self, **kwargs):
|
|
user = request.env.user
|
|
self._ensure_portal_right(user, 'portal_right_graphic_examples')
|
|
|
|
company_name = (kwargs.get('company') or '').strip()
|
|
selected_address = (kwargs.get('selected_address') or '').strip()
|
|
template_id = int(kwargs.get('template_id') or 0)
|
|
if not company_name:
|
|
_logger.info('No company name provided in graphic example start request')
|
|
return self._json_response({'ok': False, 'error': 'Bitte eine Firma eingeben.'}, status=400)
|
|
|
|
settings = self._get_adportal_settings()
|
|
template = None
|
|
if template_id:
|
|
template = self._get_visible_graphic_templates().filtered(lambda item: item.id == template_id)[:1]
|
|
template = template[0] if template else None
|
|
|
|
resolution = self._resolve_graphic_resolution(template)
|
|
_logger.info('Using resolution: %s', resolution)
|
|
|
|
mediatype_prompt = (template.adportal_ki_prompt or '').strip() if template else ''
|
|
company_context, company_context_warning = self._get_company_context_from_google(settings, company_name)
|
|
|
|
prompt, chatgpt_error = self._build_graphic_prompt_with_chatgpt(
|
|
settings,
|
|
company_name,
|
|
company_address=selected_address,
|
|
resolution_label=resolution['label'],
|
|
mediatype_name=template.medianame if template else 'Standard',
|
|
mediatype_prompt=mediatype_prompt,
|
|
company_context=company_context,
|
|
)
|
|
payload = {
|
|
'mode': 'start',
|
|
'company': company_name,
|
|
'prompt': prompt,
|
|
'template_id': template.id if template else 0,
|
|
'template_name': template.medianame if template else 'Standard',
|
|
'resolution_width': resolution['width'],
|
|
'resolution_height': resolution['height'],
|
|
'resolution_label': resolution['label'],
|
|
}
|
|
result, error = self._call_graphics_engine(settings, payload)
|
|
if error:
|
|
return self._json_response({'ok': False, 'error': error}, status=502)
|
|
if chatgpt_error and (settings and settings.adportal_graphics_use_chatgpt):
|
|
payload = dict(payload)
|
|
payload['chatgpt_warning'] = chatgpt_error
|
|
|
|
image_url, image_data = self._extract_image_from_ai_result(result)
|
|
if not image_url and not image_data:
|
|
return self._json_response({'ok': False, 'error': 'Kein Bild im API Ergebnis gefunden.'}, status=502)
|
|
|
|
return self._json_response({
|
|
'ok': True,
|
|
'status': 'OK',
|
|
'company': company_name,
|
|
'selected_address': selected_address,
|
|
'prompt': prompt,
|
|
'template_id': template.id if template else 0,
|
|
'template_name': template.medianame if template else 'Standard',
|
|
'resolution_width': resolution['width'],
|
|
'resolution_height': resolution['height'],
|
|
'resolution_label': resolution['label'],
|
|
'chatgpt_warning': chatgpt_error if (settings and settings.adportal_graphics_use_chatgpt) else '',
|
|
'google_warning': company_context_warning,
|
|
'image_url': image_url,
|
|
'image_data': image_data,
|
|
})
|
|
|
|
@http.route(['/adportal/graphic-examples/validate-address'], type='http', auth='user', methods=['POST'], csrf=False)
|
|
def adportal_graphic_examples_validate_address(self, **kwargs):
|
|
user = request.env.user
|
|
self._ensure_portal_right(user, 'portal_right_graphic_examples')
|
|
|
|
company_name = (kwargs.get('company') or '').strip()
|
|
if not company_name:
|
|
return self._json_response({'ok': False, 'error': 'Bitte eine Firma eingeben.'}, status=400)
|
|
|
|
settings = self._get_adportal_settings()
|
|
is_valid, validation_error, addresses = self._validate_company_with_google(settings, company_name)
|
|
if not is_valid:
|
|
return self._json_response({'ok': False, 'error': validation_error}, status=400)
|
|
|
|
selected_address = addresses[0] if len(addresses) == 1 else {}
|
|
return self._json_response({
|
|
'ok': True,
|
|
'status': 'OK',
|
|
'message': 'Adresse erfolgreich geprueft.',
|
|
'requires_selection': len(addresses) > 1,
|
|
'addresses': addresses,
|
|
'selected_address': selected_address,
|
|
})
|
|
|
|
@http.route(['/adportal/graphic-examples/refine'], type='http', auth='user', methods=['POST'], csrf=False)
|
|
def adportal_graphic_examples_refine(self, **kwargs):
|
|
user = request.env.user
|
|
self._ensure_portal_right(user, 'portal_right_graphic_examples')
|
|
|
|
company_name = (kwargs.get('company') or '').strip()
|
|
selected_address = (kwargs.get('selected_address') or '').strip()
|
|
instruction = (kwargs.get('instruction') or '').strip()
|
|
current_image = (kwargs.get('current_image') or '').strip()
|
|
template_id = int(kwargs.get('template_id') or 0)
|
|
|
|
if not company_name:
|
|
return self._json_response({'ok': False, 'error': 'Bitte eine Firma eingeben.'}, status=400)
|
|
if not instruction:
|
|
return self._json_response({'ok': False, 'error': 'Bitte eine Aenderung eingeben.'}, status=400)
|
|
|
|
settings = self._get_adportal_settings()
|
|
template = None
|
|
if template_id:
|
|
template = self._get_visible_graphic_templates().filtered(lambda item: item.id == template_id)[:1]
|
|
template = template[0] if template else None
|
|
|
|
resolution = self._resolve_graphic_resolution(template)
|
|
mediatype_prompt = (template.adportal_ki_prompt or '').strip() if template else ''
|
|
company_context, company_context_warning = self._get_company_context_from_google(settings, company_name)
|
|
prompt, chatgpt_error = self._build_graphic_prompt_with_chatgpt(
|
|
settings,
|
|
company_name,
|
|
company_address=selected_address,
|
|
instruction=instruction,
|
|
resolution_label=resolution['label'],
|
|
mediatype_name=template.medianame if template else 'Standard',
|
|
mediatype_prompt=mediatype_prompt,
|
|
company_context=company_context,
|
|
)
|
|
payload = {
|
|
'mode': 'refine',
|
|
'company': company_name,
|
|
'prompt': prompt,
|
|
'instruction': instruction,
|
|
'current_image': current_image,
|
|
'template_id': template.id if template else 0,
|
|
'template_name': template.medianame if template else 'Standard',
|
|
'resolution_width': resolution['width'],
|
|
'resolution_height': resolution['height'],
|
|
'resolution_label': resolution['label'],
|
|
}
|
|
result, error = self._call_graphics_engine(settings, payload)
|
|
if error:
|
|
return self._json_response({'ok': False, 'error': error}, status=502)
|
|
if chatgpt_error and (settings and settings.adportal_graphics_use_chatgpt):
|
|
payload = dict(payload)
|
|
payload['chatgpt_warning'] = chatgpt_error
|
|
|
|
image_url, image_data = self._extract_image_from_ai_result(result)
|
|
if not image_url and not image_data:
|
|
return self._json_response({'ok': False, 'error': 'Kein Bild im API Ergebnis gefunden.'}, status=502)
|
|
|
|
return self._json_response({
|
|
'ok': True,
|
|
'status': 'OK',
|
|
'company': company_name,
|
|
'selected_address': selected_address,
|
|
'prompt': prompt,
|
|
'template_id': template.id if template else 0,
|
|
'template_name': template.medianame if template else 'Standard',
|
|
'resolution_width': resolution['width'],
|
|
'resolution_height': resolution['height'],
|
|
'resolution_label': resolution['label'],
|
|
'chatgpt_warning': chatgpt_error if (settings and settings.adportal_graphics_use_chatgpt) else '',
|
|
'google_warning': company_context_warning,
|
|
'image_url': image_url,
|
|
'image_data': image_data,
|
|
})
|
|
|
|
@http.route(['/adportal/contracts'], type='http', auth='user', website=True)
|
|
def adportal_contracts(self, **kwargs):
|
|
user = request.env.user
|
|
self._ensure_portal_right(user, 'portal_right_contract_overview')
|
|
|
|
contract_model = request.env['dss.contracts'].sudo()
|
|
|
|
preferred_domain = self._get_user_contract_visibility_domain(user, require_project=True)
|
|
fallback_domain = [('project', '!=', False)]
|
|
|
|
grouped = contract_model.read_group(preferred_domain, ['project'], ['project'])
|
|
if not grouped:
|
|
grouped = contract_model.read_group(fallback_domain, ['project'], ['project'])
|
|
|
|
count_by_project = {}
|
|
project_ids = []
|
|
for row in grouped:
|
|
project_tuple = row.get('project')
|
|
if not project_tuple:
|
|
continue
|
|
project_id = project_tuple[0]
|
|
count_by_project[project_id] = row.get('project_count', row.get('__count', 0))
|
|
project_ids.append(project_id)
|
|
|
|
project_records = request.env['dss.projects'].sudo().browse(project_ids)
|
|
project_cards = []
|
|
for project in sorted(project_records, key=lambda p: ((p.projectid or 0), p.id)):
|
|
project_cards.append({
|
|
'id': project.id,
|
|
'projectid': project.projectid,
|
|
'name': project.projektname or project.name or '-',
|
|
'count': count_by_project.get(project.id, 0),
|
|
})
|
|
|
|
values = {
|
|
'username': user.name,
|
|
'project_cards': project_cards,
|
|
'page_name': 'adportal_contracts',
|
|
}
|
|
return request.render('DigitalSignage.adportal_contracts', values)
|
|
|
|
@http.route(['/adportal/contracts/project/<int:project_id>'], type='http', auth='user', website=True)
|
|
def adportal_contracts_by_project(self, project_id, **kwargs):
|
|
user = request.env.user
|
|
self._ensure_portal_right(user, 'portal_right_contract_overview')
|
|
|
|
project = request.env['dss.projects'].sudo().browse(project_id)
|
|
if not project.exists():
|
|
raise NotFound()
|
|
|
|
contract_model = request.env['dss.contracts'].sudo()
|
|
preferred_domain = self._get_user_contract_visibility_domain(user, project_id=project.id)
|
|
fallback_domain = [('project', '=', project.id)]
|
|
|
|
documents = contract_model.search(
|
|
preferred_domain,
|
|
order='write_date desc, id desc',
|
|
limit=200,
|
|
)
|
|
if not documents:
|
|
documents = contract_model.search(
|
|
fallback_domain,
|
|
order='write_date desc, id desc',
|
|
limit=200,
|
|
)
|
|
|
|
computed_contract_end_by_id = {}
|
|
for document in documents:
|
|
_logger.warning('ADPORTAL: building computed_contract_end_by_id for contract id=%s', document.id)
|
|
computed_contract_end_by_id[document.id] = self._compute_contract_end_without_extension(document)
|
|
|
|
values = {
|
|
'username': user.name,
|
|
'project': project,
|
|
'documents': documents,
|
|
'computed_contract_end_by_id': computed_contract_end_by_id,
|
|
'page_name': 'adportal_contracts_project',
|
|
}
|
|
return request.render('DigitalSignage.adportal_contracts_project', values)
|
|
|
|
@http.route(['/adportal/contracts/<string:uuid>'], type='http', auth='user', website=True)
|
|
def adportal_contract_detail(self, uuid, **kwargs):
|
|
user = request.env.user
|
|
self._ensure_portal_right(user, 'portal_right_contract_detail')
|
|
can_show_media = self._user_has_portal_right(user, 'portal_right_media_view')
|
|
can_show_graphic_examples = self._user_has_portal_right(user, 'portal_right_graphic_examples')
|
|
|
|
contract_model = request.env['dss.contracts'].sudo()
|
|
contract = contract_model.search([
|
|
('uuid', '=', uuid),
|
|
], limit=1)
|
|
|
|
self._ensure_contract_portal_access(contract, contract_model, user)
|
|
|
|
media_path = kwargs.get('media_path') or ''
|
|
media_values = self._build_media_values(contract, media_path) if can_show_media else {
|
|
'contract': contract,
|
|
'media_enabled': False,
|
|
'media_entries': [],
|
|
'media_current_path': '',
|
|
'media_parent_path': '',
|
|
'media_parent_href': '/adportal/contracts/%s' % contract.uuid,
|
|
'media_panel_url': '/adportal/contracts/%s/media/panel' % contract.uuid,
|
|
'media_breadcrumbs': [],
|
|
'media_error': '',
|
|
}
|
|
|
|
values = {
|
|
'username': user.name,
|
|
'contract': contract,
|
|
'show_media': can_show_media,
|
|
'show_graphic_examples': can_show_graphic_examples,
|
|
**media_values,
|
|
'page_name': 'adportal_contract_detail',
|
|
}
|
|
return request.render('DigitalSignage.adportal_contract_detail', values)
|
|
|
|
@http.route(['/adportal/contracts/<string:uuid>/media/panel'], type='http', auth='user', website=True)
|
|
def adportal_contract_media_panel(self, uuid, **kwargs):
|
|
user = request.env.user
|
|
self._ensure_portal_right(user, 'portal_right_contract_detail')
|
|
self._ensure_portal_right(user, 'portal_right_media_view')
|
|
|
|
contract_model = request.env['dss.contracts'].sudo()
|
|
contract = contract_model.search([
|
|
('uuid', '=', uuid),
|
|
], limit=1)
|
|
self._ensure_contract_portal_access(contract, contract_model, user)
|
|
|
|
media_values = self._build_media_values(contract, kwargs.get('media_path') or '')
|
|
return request.render('DigitalSignage.adportal_contract_media_panel', media_values)
|
|
|
|
@http.route(['/adportal/contracts/<string:uuid>/media'], type='http', auth='user', website=True)
|
|
def adportal_contract_media_download(self, uuid, **kwargs):
|
|
user = request.env.user
|
|
self._ensure_portal_right(user, 'portal_right_contract_detail')
|
|
self._ensure_portal_right(user, 'portal_right_media_view')
|
|
|
|
contract_model = request.env['dss.contracts'].sudo()
|
|
contract = contract_model.search([
|
|
('uuid', '=', uuid),
|
|
], limit=1)
|
|
self._ensure_contract_portal_access(contract, contract_model, user)
|
|
|
|
requested_rel_path = self._normalize_cloud_path(unquote(kwargs.get('path') or ''))
|
|
if not requested_rel_path:
|
|
raise NotFound()
|
|
|
|
cloud_root = self._get_contract_cloud_root(contract)
|
|
if not cloud_root:
|
|
raise NotFound()
|
|
|
|
absolute_path = self._join_cloud_paths(cloud_root, requested_rel_path)
|
|
dav_base, webdav_user, webdav_password = self._get_webdav_connection_data()
|
|
if not dav_base:
|
|
raise NotFound()
|
|
|
|
try:
|
|
client = WebdavFileSystem(dav_base, auth=(webdav_user, webdav_password))
|
|
info = client.info(absolute_path)
|
|
if str(info.get('type') or '').lower() in ('directory', 'dir'):
|
|
return request.redirect('/adportal/contracts/%s?media_path=%s' % (contract.uuid, quote(requested_rel_path)))
|
|
with client.open(absolute_path, 'rb') as stream:
|
|
data = stream.read()
|
|
except Exception:
|
|
raise NotFound()
|
|
|
|
filename = absolute_path.split('/')[-1] or 'download'
|
|
mimetype = mimetypes.guess_type(filename)[0] or 'application/octet-stream'
|
|
headers = [
|
|
('Content-Type', mimetype),
|
|
('Content-Disposition', 'inline; filename="%s"' % filename),
|
|
]
|
|
return request.make_response(data, headers=headers)
|