#!/usr/bin/env python3
"""
Improved Property Management intake for SMA Radar
Sitemap discovery, CMS endpoints, service content detection
"""
import sys
import re
import time
import json
import base64
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from datetime import datetime, timezone, timedelta
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
try:
import yaml
except ImportError:
yaml = None
from signal_db import insert_signal, record_source_run
from classification.signal_classifier import score_signal
# Service-related keywords for Property Management
SERVICE_KEYWORDS = [
'property check', 'key holding', 'home watch', 'absence management',
'preventive maintenance', 'humidity check', 'mould prevention', 'ventilation',
'air conditioning maintenance', 'water leak', 'storm check', 'pool', 'garden',
'renovation', 'emergency service', 'owner service', 'arrival preparation',
'departure check', 'winter check', 'summer check', 'hauskontrolle',
'wartung', 'instandhaltung', 'service', 'pflegeservice', 'hauswart',
'schlüssel hinterlegung', 'notdienst', 'feuchtigkeitsmessung',
'schimmel untersuchung', 'luftqualität', 'klimanalyse'
]
# CMS endpoints to check
CMS_ENDPOINTS = [
# WordPress
'/wp-json/wp/v2/posts',
'/wp-json/wp/v2/pages',
'/feed/',
'/rss/',
# Generic RSS/XML
'/blog/feed/',
'/blog/rss/',
'/news/feed/',
'/articles/feed/',
'/press/feed/',
# Sitemaps
'/sitemap.xml',
'/sitemap_index.xml',
]
def get_page_content(url, timeout=15):
"""Fetch page content with proper headers"""
req = urllib.request.Request(url, headers={
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'de-DE,de;q=0.9,es;q=0.8,en;q=0.7',
})
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.read(), resp.getcode(), dict(resp.headers)
except Exception as e:
return None, 0, {}
def extract_links_from_html(html_content, base_url):
"""Extract relevant links from HTML content"""
if not html_content:
return []
try:
html = html_content.decode('utf-8', errors='ignore')
except:
return []
# Find all href attributes
href_pattern = r'href=["\']([^"\']+)["\']'
raw_links = re.findall(href_pattern, html, re.IGNORECASE)
# Process and filter links
processed_links = []
base_parsed = urllib.parse.urlparse(base_url)
base_domain = f"{base_parsed.scheme}://{base_parsed.netloc}"
for link in raw_links:
# Skip empty, javascript, mailto, tel links
if not link or link.startswith(('javascript:', 'mailto:', 'tel:', '#')):
continue
# Convert relative URLs to absolute
if link.startswith('//'):
link = f"{base_parsed.scheme}:{link}"
elif link.startswith('/'):
link = f"{base_domain}{link}"
elif not link.startswith(('http://', 'https://')):
# Relative path
link = urllib.parse.urljoin(base_url, link)
# Only include links from the same domain
try:
link_parsed = urllib.parse.urlparse(link)
if link_parsed.netloc == base_parsed.netloc:
processed_links.append(link)
except:
continue
return list(set(processed_links)) # Remove duplicates
def is_service_related(title, content=''):
"""Check if content is related to property management services"""
text = f"{title} {content}".lower()
return any(keyword in text for keyword in SERVICE_KEYWORDS)
def extract_og_image(html_content):
"""Extract OG image from HTML content"""
if not html_content:
return "", ""
try:
html = html_content.decode('utf-8', errors='ignore')
except:
return "", ""
# OG image patterns
og_patterns = [
r'<meta[^>]+property=["\']og:image["\'][^>]+content=["\']([^"\']+)["\']',
r'<meta[^>]+content=["\']([^"\']+)["\'][^>]+property=["\']og:image["\']',
r'<meta[^>]+name=["\']twitter:image["\'][^>]+content=["\']([^"\']+)["\']',
r'<meta[^>]+content=["\']([^"\']+)["\'][^>]+name=["\']twitter:image["\']',
]
for pattern in og_patterns:
match = re.search(pattern, html, re.IGNORECASE)
if match:
image_url = match.group(1).strip()
if image_url.startswith(('http://', 'https://')):
source = 'og:image' if 'og:image' in pattern.lower() else 'twitter:image'
return image_url, source
return "", ""
def check_cms_endpoints(base_url):
"""Check CMS endpoints for recent posts"""
results = []
for endpoint in CMS_ENDPOINTS[:6]: # Limit to avoid too many requests
url = urllib.parse.urljoin(base_url, endpoint)
content, status, headers = get_page_content(url, timeout=10)
if status == 200 and content:
content_type = headers.get('content-type', '').lower()
if 'json' in content_type:
# Handle JSON responses (WordPress REST API)
try:
data = json.loads(content.decode('utf-8'))
if isinstance(data, list):
for item in data[:5]: # Limit to 5 items
title = item.get('title', {}).get('rendered', '') if isinstance(item.get('title'), dict) else str(item.get('title', ''))
link = item.get('link', '')
date = item.get('date', '')
excerpt = item.get('excerpt', {}).get('rendered', '') if isinstance(item.get('excerpt'), dict) else str(item.get('excerpt', ''))
if title and link:
results.append({
'title': title,
'url': link,
'date': date,
'summary': excerpt,
'source': endpoint
})
except:
pass
elif 'xml' in content_type or 'rss' in content_type or 'atom' in content_type:
# Handle XML/RSS feeds
try:
root = ET.fromstring(content)
# Handle different feed formats
items = []
# RSS 2.0
items.extend(root.findall('.//item'))
# Atom
items.extend(root.findall('.//{http://www.w3.org/2005/Atom}entry'))
for item in items[:5]: # Limit to 5 items
title_elem = item.find('title') or item.find('./{http://www.w3.org/2005/Atom}title')
link_elem = item.find('link') or item.find('./{http://www.w3.org/2005/Atom}link')
date_elem = item.find('pubDate') or item.find('./{http://www.w3.org/2005/Atom}published') or item.find('./{http://purl.org/rss/1.0/modules/dc/}date')
desc_elem = item.find('description') or item.find('./{http://www.w3.org/2005/Atom}summary')
title = title_elem.text.strip() if title_elem is not None and title_elem.text else ''
link = ''
if link_elem is not None:
link = link_elem.get('href') or link_elem.text or ''
date = date_elem.text.strip() if date_elem is not None and date_elem.text else ''
summary = desc_elem.text.strip() if desc_elem is not None and desc_elem.text else ''
if title and link:
results.append({
'title': title,
'url': link,
'date': date,
'summary': summary,
'source': endpoint
})
except:
pass
return results
def process_sitemap(sitemap_url):
"""Process sitemap for relevant URLs"""
content, status, headers = get_page_content(sitemap_url, timeout=15)
if status != 200 or not content:
return []
content_type = headers.get('content-type', '').lower()
results = []
try:
if 'xml' in content_type:
root = ET.fromstring(content)
# Handle sitemap index
if root.tag.endswith('sitemapindex'):
sitemap_elements = root.findall('.//{*}sitemap')
for sitemap_elem in sitemap_elements[:3]: # Limit to 3 sub-sitemaps
loc_elem = sitemap_elem.find('{*}loc')
if loc_elem is not None and loc_elem.text:
sub_sitemap_url = loc_elem.text.strip()
sub_results = process_sitemap(sub_sitemap_url)
results.extend(sub_results)
# Handle URL set
elif root.tag.endswith('urlset'):
url_elements = root.findall('.//{*}url')
for url_elem in url_elements[:20]: # Limit to 20 URLs per sitemap
loc_elem = url_elem.find('{*}loc')
lastmod_elem = url_elem.find('{*}lastmod')
if loc_elem is not None and loc_elem.text:
url = loc_elem.text.strip()
lastmod = lastmod_elem.text.strip() if lastmod_elem is not None and lastmod_elem.text else ''
# Check if URL looks relevant to property management services
if any(keyword in url.lower() for keyword in [
'blog', 'news', 'article', 'service', 'wartung', 'pflege',
'check', 'inspection', 'wartung', 'service', 'notfall',
'feuchte', 'schimmel', 'luft', 'klima', 'pool', 'garten'
]):
results.append({
'title': url.split('/')[-1].replace('-', ' ').replace('_', ' ').title(),
'url': url,
'date': lastmod,
'summary': '',
'source': 'sitemap'
})
except:
pass
return results
def entity_to_signal(entity, content_info):
"""Convert content info to signal format"""
text = f"{content_info['title']} {content_info.get('summary', '')}"
sd = score_signal(text, language="de", base_category="feuchte",
source_url=content_info['url'])
# Boost score for service-related content
service_bonus = 15 if is_service_related(content_info['title'], content_info.get('summary', '')) else 0
final_score = min(sd['radar_score'] + service_bonus, 100)
if final_score < 31:
return None
expires = (datetime.now(timezone.utc) + timedelta(hours=336)).isoformat()
# Extract image
image_url, image_source = "", ""
if content_info.get('url'):
# Try to get image from the page
page_content, _, _ = get_page_content(content_info['url'], timeout=10)
if page_content:
image_url, image_source = extract_og_image(page_content)
signal = {
'source_type': 'watch_entity',
'source_platform': 'web',
'external_id': '',
'observed_at': content_info.get('date', datetime.now(timezone.utc).isoformat()),
'source_name': f"PM-IMPROVED: {entity['canonical_name']} ({content_info['source']})",
'source_url': content_info['url'],
'language': (entity.get('languages') or ['de'])[0],
'signal_category': sd['signal_category'] or 'property_service',
'topic': content_info['title'][:200],
'short_summary': (content_info.get('summary', '') or content_info['title'])[:500],
'extracted_hook': sd['extracted_hook'],
'emotional_direction': sd['emotional_direction'],
'urgency_level': sd['urgency_level'],
'mallorca_relevance': sd['mallorca_relevance'],
'risk_relevance': sd['risk_relevance'],
'estimated_noise_level': sd['estimated_noise_level'],
'suggested_case_types': sd['suggested_case_types'],
'suggested_platforms': sd['suggested_platforms'],
'suggested_cta': sd['suggested_cta'],
'confidence_score': sd['confidence_score'],
'radar_score': final_score,
'decay_rate': 0.8,
'expires_at': expires,
'virality_level': sd['virality_level'],
'emotionality_level': sd['emotionality_level'],
'comment_potential': sd['comment_potential'],
'content_potential': sd['content_potential'],
'score_breakdown': sd['score_breakdown'],
'platform_facebook': 1,
'topic_class': 'sozial',
'retrieval_mode': 'sitemap' if 'sitemap' in content_info.get('source', '') else 'cms_endpoint',
'watch_entity_id': entity['entity_id'],
'watch_sector': entity['sector'],
'watch_priority': entity.get('priority', 'normal'),
'image_url': image_url,
'image_source': image_source,
'image_found': 1 if image_url else 0,
}
return signal
def run_improved_property_management(cfg):
"""Run improved Property Management intake"""
# For now, we'll integrate this into the watch intake
# This function shows the approach but the actual implementation
# will be in the updated watch_intake.py
print("[PM-IMPROVED] Improved Property Management intake ready for integration")
return 0
if __name__ == '__main__':
# Test standalone
cfg = {}
run_improved_property_management(cfg)