Rename main.py to scraper.py

main.py

@@ -1,305 +0,0 @@
-import os
-from typing import List, Dict, Optional
-from datetime import datetime, timedelta
-
-from fastapi import FastAPI, HTTPException, Depends, WebSocket, WebSocketDisconnect, status
-from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
-from fastapi.responses import HTMLResponse
-from pydantic import BaseModel
-from sqlalchemy import create_engine, Column, Integer, String, Table, ForeignKey, DateTime, Text
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import sessionmaker, relationship, Session
-import uvicorn
-
-# امنیت: JWT و رمزنگاری پسوردها
-from jose import JWTError, jwt
-from passlib.context import CryptContext
-
-# پیکربندی
-SECRET_KEY = "your-secret-key"  # در محصولات واقعی تعویض شود و از متغیرهای محیطی استفاده کنید
-ALGORITHM = "HS256"
-ACCESS_TOKEN_EXPIRE_MINUTES = 30
-
-# تنظیمات رمزنگاری رمز عبور
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
-
-# ایجاد پایگاه داده SQLite
-DATABASE_URL = "sqlite:///./chat.db"
-engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
-SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
-Base = declarative_base()
-
-# جداول پایگاه‌داده
-class UserModel(Base):
-    __tablename__ = "users"
-    id = Column(Integer, primary_key=True, index=True)
-    username = Column(String, unique=True, index=True)
-    hashed_password = Column(String)
-
-# جدول گروه‌ها
-class GroupModel(Base):
-    __tablename__ = "groups"
-    id = Column(Integer, primary_key=True, index=True)
-    group_name = Column(String, unique=True, index=True)
-    owner_id = Column(Integer, ForeignKey("users.id"))
-    owner = relationship("UserModel")
-    
-# جدول پیام‌ها
-class MessageModel(Base):
-    __tablename__ = "messages"
-    id = Column(Integer, primary_key=True, index=True)
-    sender_id = Column(Integer, ForeignKey("users.id"))
-    group_id = Column(Integer, ForeignKey("groups.id"))
-    content = Column(Text)
-    timestamp = Column(DateTime, default=datetime.utcnow)
-
-# ایجاد جداول اگر وجود نداشتند
-Base.metadata.create_all(bind=engine)
-
-# مدل‌های Pydantic
-class User(BaseModel):
-    username: str
-    password: str
-
-class UserOut(BaseModel):
-    id: int
-    username: str
-    class Config:
-        orm_mode = True
-
-class Token(BaseModel):
-    access_token: str
-    token_type: str
-
-class Group(BaseModel):
-    group_name: str
-
-class GroupOut(BaseModel):
-    id: int
-    group_name: str
-    owner: UserOut
-    class Config:
-        orm_mode = True
-
-class Message(BaseModel):
-    sender: str
-    content: str
-    timestamp: datetime
-
-# توابع مربوط به رمزنگاری
-def verify_password(plain_password, hashed_password):
-    return pwd_context.verify(plain_password, hashed_password)
-
-def get_password_hash(password):
-    return pwd_context.hash(password)
-
-# توابع مربوط به DB
-def get_db():
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
-
-# توابع احراز هویت JWT
-def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
-    to_encode = data.copy()
-    if expires_delta:
-        expire = datetime.utcnow() + expires_delta
-    else:
-        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
-    to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-    return encoded_jwt
-
-def get_user(db: Session, username: str):
-    return db.query(UserModel).filter(UserModel.username == username).first()
-
-def authenticate_user(db: Session, username: str, password: str):
-    user = get_user(db, username)
-    if not user:
-        return False
-    if not verify_password(password, user.hashed_password):
-        return False
-    return user
-
-async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
-    credentials_exception = HTTPException(
-        status_code=status.HTTP_401_UNAUTHORIZED,
-        detail="Could not validate credentials",
-        headers={"WWW-Authenticate": "Bearer"},
-    )
-    try:
-        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
-        username: str = payload.get("sub")
-        if username is None:
-            raise credentials_exception
-    except JWTError:
-        raise credentials_exception
-        
-    user = get_user(db, username=username)
-    if user is None:
-        raise credentials_exception
-    return user
-
-# ایجاد سرور FastAPI
-app = FastAPI()
-
-# لاگ ساده: در پروژه‌های عملی می‌توانید از logging پیشرفته استفاده کنید
-@app.middleware("http")
-async def add_process_time_header(request, call_next):
-    response = await call_next(request)
-    return response
-
-# --- Endpointهای احراز هویت ---
-
-@app.post("/register/", response_model=UserOut)
-def register_user(user: User, db: Session = Depends(get_db)):
-    db_user = get_user(db, username=user.username)
-    if db_user:
-        raise HTTPException(status_code=400, detail="Username already registered")
-    hashed_pwd = get_password_hash(user.password)
-    new_user = UserModel(username=user.username, hashed_password=hashed_pwd)
-    db.add(new_user)
-    db.commit()
-    db.refresh(new_user)
-    return new_user
-
-@app.post("/token", response_model=Token)
-def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), 
-                           db: Session = Depends(get_db)):
-    user = authenticate_user(db, form_data.username, form_data.password)
-    if not user:
-        raise HTTPException(status_code=401, detail="Incorrect username or password")
-    
-    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
-    access_token = create_access_token(
-        data={"sub": user.username}, expires_delta=access_token_expires
-    )
-    return {"access_token": access_token, "token_type": "bearer"}
-
-# --- Endpoint های مربوط به گروه‌ها ---
-
-@app.post("/groups/", response_model=GroupOut)
-def create_group(group: Group, current_user: UserModel = Depends(get_current_user),
-                 db: Session = Depends(get_db)):
-    # بررسی وجود گروه با نام مشابه
-    db_group = db.query(GroupModel).filter(GroupModel.group_name == group.group_name).first()
-    if db_group:
-        raise HTTPException(status_code=400, detail="Group name already exists")
-    new_group = GroupModel(group_name=group.group_name, owner_id=current_user.id)
-    db.add(new_group)
-    db.commit()
-    db.refresh(new_group)
-    return new_group
-
-@app.get("/groups/", response_model=List[GroupOut])
-def get_groups(current_user: UserModel = Depends(get_current_user),
-               db: Session = Depends(get_db)):
-    # فرض می‌کنیم هر گروهی متعلق به یک کاربر نیست، 
-    # در دنیای واقعی ممکن است کاربران عضو گروه‌های مختلفی باشند (نیاز به جدول ارتباطی دارد)
-    groups = db.query(GroupModel).all()
-    return groups
-
-# --- WebSocket برای ارسال و دریافت پیام‌ها ---
-# در این مثال از یک ConnectionManager ساده استفاده می‌کنیم
-class ConnectionManager:
-    def __init__(self):
-        # نگهداری اتصالات بر اساس id گروه
-        self.active_connections: Dict[int, List[WebSocket]] = {}
-
-    async def connect(self, group_id: int, websocket: WebSocket):
-        await websocket.accept()
-        if group_id not in self.active_connections:
-            self.active_connections[group_id] = []
-        self.active_connections[group_id].append(websocket)
-
-    def disconnect(self, group_id: int, websocket: WebSocket):
-        if group_id in self.active_connections:
-            self.active_connections[group_id].remove(websocket)
-
-    async def broadcast(self, group_id: int, message: Message):
-        if group_id in self.active_connections:
-            for connection in self.active_connections[group_id]:
-                await connection.send_json(message.dict())
-
-manager = ConnectionManager()
-
-@app.websocket("/ws/groups/{group_id}")
-async def websocket_endpoint(websocket: WebSocket, group_id: int,
-                             token: str = Depends(oauth2_scheme),
-                             db: Session = Depends(get_db)):
-    # احراز هویت مجدد روی WebSocket
-    try:
-        current_user = await get_current_user(token, db)
-    except HTTPException:
-        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
-        return
-    
-    await manager.connect(group_id, websocket)
-    try:
-        while True:
-            data = await websocket.receive_json()
-            # ایجاد شی پیام جهت ارسال
-            message = Message(
-                sender=current_user.username,
-                content=data.get("content"),
-                timestamp=datetime.utcnow()
-            )
-            # ذخیره پیام در پایگاه داده
-            new_msg = MessageModel(sender_id=current_user.id, group_id=group_id, content=message.content)
-            db.add(new_msg)
-            db.commit()
-            await manager.broadcast(group_id, message)
-    except WebSocketDisconnect:
-        manager.disconnect(group_id, websocket)
-    except Exception as e:
-        # در صورت بروز خطای دیگر، می‌توانیم آن را لاگ کرده و websocket را ببندیم.
-        await websocket.close()
-
-# نمونه از صفحه HTML جهت تست WebSocket (در پروژه تولید رابط کاربری جداگانه نوشته می شود)
-html = """
-<!DOCTYPE html>
-<html>
-    <head>
-        <title>WebSocket Chat Room</title>
-    </head>
-    <body>
-        <h1>WebSocket Chat</h1>
-        <form id="form">
-            <input type="text" id="messageText" autocomplete="off"/>
-            <button>Send</button>
-        </form>
-        <ul id='messages'>
-        </ul>
-        <script>
-            // برای تست، جای token و group_id را تنظیم کنید
-            let token = prompt("Enter your JWT token:");
-            let group_id = prompt("Enter group id:");
-            const ws = new WebSocket(`ws://localhost:8000/ws/groups/${group_id}?token=${token}`);
-            ws.onmessage = function(event) {
-                const messages = document.getElementById('messages')
-                const message = document.createElement('li')
-                const data = JSON.parse(event.data)
-                message.innerText = data.sender + ": " + data.content;
-                messages.appendChild(message);
-            };
-            document.getElementById('form').onsubmit = function(e) {
-                e.preventDefault()
-                const input = document.getElementById("messageText")
-                ws.send(JSON.stringify({content: input.value}))
-                input.value = ''
-            }
-        </script>
-    </body>
-</html>
-"""
-
-@app.get("/")
-def get():
-    return HTMLResponse(html)
-
-# اجرای سرور
-if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=8000)

scraper.py

@@ -0,0 +1,46 @@
+from selenium import webdriver
+from selenium.webdriver.chrome.service import Service
+from webdriver_manager.chrome import ChromeDriverManager
+from selenium.webdriver.chrome.options import Options
+import zipfile
+import os
+
+def scrape_pages(base_url, start_page, end_page):
+    chrome_options = Options()
+    chrome_options.add_argument("--headless")
+    chrome_options.add_argument("--no-sandbox")
+    chrome_options.add_argument("--disable-dev-shm-usage")
+
+    driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=chrome_options)
+
+    if not os.path.exists('pages'):
+        os.makedirs('pages')
+
+    for page in range(start_page, end_page + 1):
+        url = f"{base_url}?page={page}"
+        driver.get(url)
+        html = driver.page_source
+
+        with open(f'pages/t{page}.html', 'w', encoding='utf-8') as f:
+            f.write(html)
+        print(f"Saved: t{page}.html")
+
+    driver.quit()
+
+    # ایجاد فایل ZIP
+    with zipfile.ZipFile('html.zip', 'w', zipfile.ZIP_DEFLATED) as zipf:
+        for root, dirs, files in os.walk('pages'):
+            for file in files:
+                zipf.write(os.path.join(root, file), file)
+
+    print("All pages saved and zipped as html.zip")
+
+if __name__ == "__main__":
+    base_url = "https://shahvani.com/dastans"
+    start_page = 1
+    end_page = 10  # این عدد را می‌توانید به تابع بدهید یا از کاربر بگیرید
+
+    # مثال: اگر می‌خواهید تعداد صفحات را به صورت پویا مشخص کنید
+    # end_page = get_last_page_number()  # تابعی که تعداد صفحات را برمی‌گرداند
+
+    scrape_pages(base_url, start_page, end_page)