# -*- 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/'], 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/'], 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//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//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)