Dive into Code

def get_obj(obj):
    from django.db import models
    app_label = obj._meta.app_label
    print(app_label)
    fields = obj._meta.get_fields()
    for field in fields:
        value = getattr(obj,field.name)
        if isinstance(field,models.ForeignKey):
            print(field.remote_field.model)
        elif isinstance(value,models.Manager):
            related_objects = value.get_queryset()
            for r_obj in related_objects:
                print(r_obj)
        else:
            print("name==>",field.name, "value==>",value)
  
model_obj = Model.objects.get(id=1)
get_obj(model_obj)

"""
Async Wikipedia crawler with article-only link filter
====================================================
Script by parsing *infobox* tables **and** restricting the
link queue to real article pages whose paths match the classic “/wiki/Title”
pattern (no `File:`, `Help:`, etc.).

Run with:
    python async_crawler_infobox.py

Dependencies (pip install ...): aiohttp, lxml, beautifulsoup4, sqlalchemy, psycopg2-binary
"""

import asyncio
import aiohttp
import json
import re
from lxml import html
from urllib.parse import urljoin, urlparse

from bs4 import BeautifulSoup

from sqlalchemy import (
    create_engine,
    Column,
    Integer,
    String,
    Text,
    ForeignKey,
    Table,
)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# ───── Config ─────
DATABASE_URL = "postgresql+psycopg2://username:password@localhost:5432/pages"
MAX_CONCURRENCY = 10

# only follow links like /wiki/Foo (exclude namespaces containing ':' and non-article paths)
WIKI_PATH_PATTERN = re.compile(r"^/wiki/[^:]+$")

# ───── Database Setup ─────
engine = create_engine(DATABASE_URL, echo=False)
Session = sessionmaker(bind=engine)
session = Session()
Base = declarative_base()

page_children = Table(
    "page_children",
    Base.metadata,
    Column("parent_id", Integer, ForeignKey("pages.id", ondelete="CASCADE"), primary_key=True),
    Column("child_id", Integer, ForeignKey("pages.id", ondelete="CASCADE"), primary_key=True),
)


def extract_infobox(content: bytes) -> dict:
    """Return key→value pairs from the first Wikipedia infobox on the page."""
    soup = BeautifulSoup(content, "lxml")
    table = soup.select_one("table.infobox")
    if not table:
        return {}

    records = []
    for tr in table.select("tr"):
        hdr = tr.find("th", attrs={"scope": "row"})
        if hdr is None:
            continue
        key = hdr.get_text(" ", strip=True)
        td = tr.find("td")
        if not td:
            continue
        for sup in td.select("sup, .reference"):
            sup.decompose()
        val = td.get_text(" ", strip=True)
        records.append((key, val))
    return dict(records)


class Page(Base):
    __tablename__ = "pages"

    id = Column(Integer, primary_key=True)
    url = Column(String, unique=True, nullable=False)
    title = Column(String)
    description = Column(String)
    keywords = Column(String)
    infobox = Column(Text)  # JSON-encoded string

    children = relationship(
        "Page",
        secondary=page_children,
        primaryjoin=id == page_children.c.parent_id,
        secondaryjoin=id == page_children.c.child_id,
        backref="parents",
    )


Base.metadata.create_all(engine)

# ───── Async Crawler ─────
visited: set[str] = set()
semaphore = asyncio.Semaphore(MAX_CONCURRENCY)


async def fetch_and_parse(session_http: aiohttp.ClientSession, url: str, base_domain: str, depth: int | None = None):
    """Fetch *url*, store metadata + infobox, then crawl its article children."""

    # stop conditions
    if url in visited or urlparse(url).netloc != base_domain:
        return
    if depth is not None and depth < 0:
        return

    visited.add(url)
    print(f"[+] Crawling: {url} (depth={'∞' if depth is None else depth})")

    # ── Fetch ──
    try:
        async with semaphore, session_http.get(url, timeout=10) as resp:
            if resp.status != 200:
                return
            content = await resp.read()
    except Exception as e:
        print(f"[-] Error fetching {url}: {e}")
        return

    tree = html.fromstring(content)

    # ── Extract metadata ──
    title_text = tree.findtext(".//title") or ""
    meta_desc = tree.xpath("//meta[@name='description']/@content")
    meta_keywords = tree.xpath("//meta[@name='keywords']/@content")

    # ── Extract infobox ──
    infobox_data = extract_infobox(content)

    page_obj = session.query(Page).filter_by(url=url).first()
    if not page_obj:
        page_obj = Page(url=url)
        session.add(page_obj)

    page_obj.title = title_text.strip()
    page_obj.description = meta_desc[0].strip() if meta_desc else ""
    page_obj.keywords = meta_keywords[0].strip() if meta_keywords else ""
    page_obj.infobox = json.dumps(infobox_data, ensure_ascii=False) if infobox_data else None

    session.commit()

    # ── Recurse ──
    tasks = []
    for el, attr, link, _ in tree.iterlinks():
        if el.tag != "a" or attr != "href":
            continue
        child_url = urljoin(url, link).split("#")[0].rstrip("/")

        # early rejections
        if child_url.startswith(("mailto:", "javascript:")) or child_url in visited:
            continue
        if urlparse(child_url).netloc != base_domain:
            continue
        if not WIKI_PATH_PATTERN.match(urlparse(child_url).path):
            continue  # skip non-article paths

        child_page = session.query(Page).filter_by(url=child_url).first()
        if not child_page:
            child_page = Page(url=child_url)
            session.add(child_page)
            session.commit()

        if child_page not in page_obj.children:
            page_obj.children.append(child_page)

        next_depth = None if depth is None else depth - 1
        tasks.append(fetch_and_parse(session_http, child_url, base_domain, next_depth))

    session.commit()
    if tasks:
        await asyncio.gather(*tasks)


async def crawl(start_url: str, depth: int | None = None):
    """Entry point: crawl starting from *start_url* down to *depth*."""
    base_domain = urlparse(start_url).netloc
    async with aiohttp.ClientSession(headers={"User-Agent": "Mozilla/5.0"}) as session_http:
        await fetch_and_parse(session_http, start_url, base_domain, depth)


# ───── Export to JSON ─────

def export_to_json(filename: str = "pages_export.json"):
    rows = []
    for p in session.query(Page).all():
        rows.append(
            {
                "url": p.url,
                "title": p.title,
                "description": p.description,
                "keywords": p.keywords,
                "infobox": json.loads(p.infobox) if p.infobox else {},
                "children": [c.url for c in p.children],
            }
        )
    with open(filename, "w", encoding="utf-8") as fh:
        json.dump(rows, fh, ensure_ascii=False, indent=2)
    print(f"[✓] Exported to {filename}")


# ───── Main Entrypoint ─────
if __name__ == "__main__":
    seed = ""
    asyncio.run(crawl(seed, depth=None))
    export_to_json()

from collections import OrderedDict


def flatten_dict_keep_lists(data):
    flat = {}

    def _flatten(obj, parent_key=''):
        if isinstance(obj, (dict, OrderedDict)):
            for k, v in obj.items():
                key = k.split(':')[-1] if ':' in k else k
                _flatten(v, key)
        elif isinstance(obj, list):
            # If it's a list of dicts, flatten each item
            if all(isinstance(i, (dict, OrderedDict)) for i in obj):
                flat[parent_key] = [
                    flatten_dict_keep_lists(i) for i in obj
                ]
            else:
                # List of values, store as is
                flat[parent_key] = obj
        else:
            flat[parent_key] = obj

    _flatten(data)
    return flat

def unique_items(keys, objects):
    # Extract relevant items based on keys and create frozensets for uniqueness
    items = [{key: obj[key] for key in keys if key in obj} for obj in objects]
    # Use frozenset to ensure uniqueness
    set_of_dicts = {frozenset(d.items()) for d in items}
    # Convert back to dictionaries for easier usability
    unique_dicts = [dict(s) for s in set_of_dicts]
    print(len(unique_dicts))
    return unique_dicts

from django import forms
from django.forms import ModelForm

def create_modelform(model_class, include_fields=None, exclude_fields=None):
    """
    Dynamically create a ModelForm based on the model class passed as an argument.
    You can specify fields to include or exclude.
    :param model_class: The model class to base the form on.
    :param include_fields: A list of fields to include in the form (default: all fields).
    :param exclude_fields: A list of fields to exclude from the form (default: None).
    """
    class DynamicModelForm(ModelForm):
        class Meta:
            model = model_class
            # If specific fields are provided, use them; otherwise, include all fields
            if include_fields is not None:
                fields = include_fields
            elif exclude_fields is not None:
                exclude = exclude_fields
            else:
                fields = '__all__'  # Include all fields by default

    return DynamicModelForm