Ipcam Telegram Group Info

logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO ) logger = logging.getLogger(name)

class IPCameraTelegramMonitor: """ Monitor IP cameras through Telegram group with features like: - Motion detection - Snapshot capture - Recording management - Face detection (optional) """

def __init__(self, config_path: str = "config.json"):
    """
    Initialize the IP camera Telegram monitor.
Args:
        config_path: Path to configuration file
    """
    self.config = self.load_config(config_path)
    self.bot_token = self.config["telegram_bot_token"]
    self.allowed_group_id = self.config["telegram_group_id"]
    self.ip_cameras = self.config["ip_cameras"]
# Storage paths
    self.media_path = Path(self.config.get("media_storage_path", "./camera_media"))
    self.media_path.mkdir(exist_ok=True)
# Motion detection settings
    self.motion_threshold = self.config.get("motion_threshold", 25)
    self.motion_history = {}
def load_config(self, config_path: str) -> Dict:
    """Load configuration from JSON file."""
    default_config = 
        "telegram_bot_token": "YOUR_BOT_TOKEN",
        "telegram_group_id": -1001234567890,
        "media_storage_path": "./camera_media",
        "motion_threshold": 25,
        "ip_cameras": [
"name": "Front Door",
                "url": "http://192.168.1.100/video",
                "snapshot_url": "http://192.168.1.100/snapshot",
                "enabled": True
            ,
"name": "Backyard",
                "url": "http://192.168.1.101/video",
                "snapshot_url": "http://192.168.1.101/snapshot",
                "enabled": True
]
try:
        with open(config_path, 'r') as f:
            config = json.load(f)
            # Merge with defaults for missing keys
            for key, value in default_config.items():
                if key not in config:
                    config[key] = value
            return config
    except FileNotFoundError:
        logger.warning(f"Config file config_path not found. Using defaults.")
        return default_config
async def handle_group_messages(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle messages from the Telegram group."""
    if not update.message or not update.message.chat:
        return
chat_id = update.message.chat.id
# Check if message is from allowed group
    if chat_id != self.allowed_group_id:
        return
message_text = update.message.text if update.message.text else ""
    message_lower = message_text.lower()
# Handle commands from group
    if message_text.startswith('/'):
        await self.handle_commands(update, context)
    elif any(cmd in message_lower for cmd in ['photo', 'snap', 'capture']):
        await self.capture_and_send_snapshot(update, context)
    elif any(cmd in message_lower for cmd in ['video', 'record', 'stream']):
        await self.start_video_preview(update, context)
    elif 'status' in message_lower:
        await self.send_status(update, context)
    elif 'help' in message_lower:
        await self.send_help(update, context)
async def handle_commands(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle slash commands."""
    commands = 
        '/start': self.start_command,
        '/help': self.help_command,
        '/snapshot': self.snapshot_command,
        '/status': self.status_command,
        '/record': self.record_command,
        '/motion_on': self.enable_motion_detection,
        '/motion_off': self.disable_motion_detection,
        '/settings': self.show_settings
if update.message.text:
        cmd = update.message.text.split()[0].lower()
        if cmd in commands:
            await commands[cmd](update, context)
async def start_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle /start command."""
    await update.message.reply_text(
        "🎥 IP Camera Monitor Active!\n\n"
        "Available commands:\n"
        "/snapshot - Take a snapshot from all cameras\n"
        "/status - Check camera status\n"
        "/record - Start/stop recording\n"
        "/motion_on - Enable motion detection\n"
        "/motion_off - Disable motion detection\n"
        "/help - Show all commands"
    )
async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Show help information."""
    help_text = """

🤖 IP Camera Bot Commands

Camera Controls: • /snapshot [camera_name] - Take a snapshot • /status - Check camera status • /record [duration] - Record video (duration in seconds)

Motion Detection: • /motion_on - Enable motion detection • /motion_off - Disable motion detection

Settings: • /settings - View current settings

Examples: • /snapshot Front Door - Capture from specific camera • /record 30 - Record for 30 seconds • /motion_on - Start motion detection ipcam telegram group

Reply with camera names or leave blank for all cameras """ await update.message.reply_text(help_text, parse_mode='Markdown')

async def snapshot_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle /snapshot command."""
    args = context.args
    camera_name = ' '.join(args) if args else None
await update.message.reply_text("📸 Capturing snapshot...")
if camera_name:
        # Capture from specific camera
        camera = next((c for c in self.ip_cameras if c['name'].lower() == camera_name.lower()), None)
        if camera and camera['enabled']:
            await self.capture_and_send_camera(update, camera)
        else:
            await update.message.reply_text(f"❌ Camera 'camera_name' not found or disabled")
    else:
        # Capture from all cameras
        tasks = [self.capture_and_send_camera(update, camera) 
                for camera in self.ip_cameras if camera['enabled']]
        await asyncio.gather(*tasks)
async def capture_and_send_camera(self, update: Update, camera: Dict):
    """Capture and send snapshot from a specific camera."""
    try:
        snapshot_url = camera.get('snapshot_url')
        if not snapshot_url:
            await update.message.reply_text(f"❌ camera['name']: No snapshot URL configured")
            return
# Fetch image from camera
        response = requests.get(snapshot_url, timeout=10)
        response.raise_for_status()
# Save snapshot locally
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"camera['name']_timestamp.jpg"
        filepath = self.media_path / filename
with open(filepath, 'wb') as f:
            f.write(response.content)
# Send to Telegram
        await update.message.reply_photo(
            photo=open(filepath, 'rb'),
            caption=f"📷 camera['name']\n🕐 datetime.now().strftime('%Y-%m-%d %H:%M:%S')"
        )
logger.info(f"Snapshot captured from camera['name']")
except requests.RequestException as e:
        logger.error(f"Failed to capture from camera['name']: e")
        await update.message.reply_text(f"❌ camera['name']: Failed to capture snapshot")
async def capture_and_send_snapshot(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Capture and send snapshots from all enabled cameras."""
    await self.snapshot_command(update, context)
async def status_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Send camera status."""
    await self.send_status(update, context)
async def send_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Send detailed status of all cameras."""
    status_text = "📊 **Camera Status**\n\n"
for camera in self.ip_cameras:
        status_icon = "✅" if camera['enabled'] else "❌"
        status_text += f"status_icon **camera['name']**\n"
        status_text += f"   URL: camera['url']\n"
        status_text += f"   Enabled: camera['enabled']\n\n"
status_text += f"📁 Storage: self.media_path.absolute()\n"
    status_text += f"🎯 Motion Threshold: self.motion_threshold"
await update.message.reply_text(status_text, parse_mode='Markdown')
async def record_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle video recording command."""
    args = context.args
    duration = int(args[0]) if args and args[0].isdigit() else 10
await update.message.reply_text(f"🎥 Recording started for duration seconds...")
# Record from all cameras
    for camera in self.ip_cameras:
        if camera['enabled']:
            asyncio.create_task(self.record_from_camera(update, camera, duration))
async def record_from_camera(self, update: Update, camera: Dict, duration: int):
    """Record video from a camera stream."""
    try:
        cap = cv2.VideoCapture(camera['url'])
        if not cap.isOpened():
            await update.message.reply_text(f"❌ camera['name']: Cannot access stream")
            return
# Setup video writer
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"camera['name']_timestamp.mp4"
        filepath = self.media_path / filename
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        fps = 20
        frame_width = int(cap.get(3))
        frame_height = int(cap.get(4))
        out = cv2.VideoWriter(str(filepath), fourcc, fps, (frame_width, frame_height))
# Record for specified duration
        start_time = asyncio.get_event_loop().time()
        frames_written = 0
while (asyncio.get_event_loop().time() - start_time) < duration:
            ret, frame = cap.read()
            if ret:
                out.write(frame)
                frames_written += 1
            await asyncio.sleep(1/fps)
cap.release()
        out.release()
# Send video to Telegram
        if filepath.exists() and filepath.stat().st_size > 0:
            with open(filepath, 'rb') as video_file:
                await update.message.reply_video(
                    video=video_file,
                    caption=f"🎬 camera['name']\n⏱️ Duration: durations\n📅 datetime.now().strftime('%Y-%m-%d %H:%M:%S')"
                )
            logger.info(f"Recording saved from camera['name']: frames_written frames")
        else:
            await update.message.reply_text(f"❌ camera['name']: Recording failed")
except Exception as e:
        logger.error(f"Recording error for camera['name']: e")
        await update.message.reply_text(f"❌ camera['name']: Recording error")
async def enable_motion_detection(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Enable motion detection for cameras."""
    # Start motion detection tasks for each camera
    self.motion_detection_tasks = []
    for camera in self.ip_cameras:
        if camera['enabled']:
            task = asyncio.create_task(self.motion_detection_loop(update, camera))
            self.motion_detection_tasks.append(task)
await update.message.reply_text("🔍 Motion detection enabled for all cameras")
async def disable_motion_detection(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Disable motion detection."""
    if hasattr(self, 'motion_detection_tasks'):
        for task in self.motion_detection_tasks:
            task.cancel()
        await update.message.reply_text("🔍 Motion detection disabled")
async def motion_detection_loop(self, update: Update, camera: Dict):
    """Continuous motion detection loop for a camera."""
    cap = cv2.VideoCapture(camera['url'])
    if not cap.isOpened():
        logger.error(f"Cannot open stream for camera['name']")
        return
# Initialize background subtractor
    fgbg = cv2.createBackgroundSubtractorMOG2()
    last_motion_time = 0
    cooldown_seconds = 10  # Avoid spam
try:
        while True:
            ret, frame = cap.read()
            if not ret:
                await asyncio.sleep(1)
                continue
# Apply background subtraction
            fgmask = fgbg.apply(frame)
# Calculate motion area
            motion_area = np.sum(fgmask > 0)
            frame_area = frame.shape[0] * frame.shape[1]
            motion_percentage = (motion_area / frame_area) * 100
# Check if motion detected
            current_time = asyncio.get_event_loop().time()
            if motion_percentage > self.motion_threshold and (current_time - last_motion_time) > cooldown_seconds:
                last_motion_time = current_time
# Save motion frame
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                filename = f"motion_camera['name']_timestamp.jpg"
                filepath = self.media_path / filename
                cv2.imwrite(str(filepath), frame)
# Send alert to Telegram
                await update.message.reply_photo(
                    photo=open(filepath, 'rb'),
                    caption=f"⚠️ Motion detected on camera['name']!\n"
                           f"📊 Motion: motion_percentage:.1f%\n"
                           f"🕐 datetime.now().strftime('%Y-%m-%d %H:%M:%S')"
                )
logger.info(f"Motion detected on camera['name']: motion_percentage:.1f%")
await asyncio.sleep(0.1)  # Check every 100ms
except asyncio.CancelledError:
        logger.info(f"Motion detection stopped for camera['name']")
    finally:
        cap.release()
async def show_settings(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Show current settings."""
    settings_text = """

⚙️ Current Settings

Motion Detection: • Threshold: {}

Storage: • Path: {}

Cameras: • Total: {} • Enabled: {}/{} """.strip().format( self.motion_threshold, self.media_path, len(self.ip_cameras), sum(1 for c in self.ip_cameras if c['enabled']), len(self.ip_cameras) )

    await update.message.reply_text(settings_text, parse_mode='Markdown')
async def run(self):
    """Main function to run the bot."""
    # Create application
    application = Application.builder().token(self.bot_token).build()
# Add handlers
    application.add_handler(MessageHandler(
        filters.TEXT & ~filters.COMMAND,
        self.handle_group_messages
    ))
    application.add_handler(CommandHandler('start', self.start_command))
    application.add_handler(CommandHandler('help', self.help_command))
    application.add_handler(CommandHandler('snapshot', self.snapshot_command))
    application.add_handler(CommandHandler('status', self.status_command))
    application.add_handler(CommandHandler('record', self.record_command))
    application.add_handler(CommandHandler('motion_on', self.enable_motion_detection))
    application.add_handler(CommandHandler('motion_off', self.disable_motion_detection))
    application.add_handler(CommandHandler('settings', self.show_settings))
# Start the bot
    logger.info("Starting IP Camera Telegram Monitor...")
    await application.initialize()
    await application.start()
    await application.updater.start_polling()
# Keep running
    try:
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        logger.info("Shutting down...")
        await application.updater.stop()
        await application.stop()
        await application.shutdown()

The concept is simple, yet terrifying. Internet Protocol (IP) cameras—those sleek, white domes and orbs sold by Amazon Ring, Nest, Wyze, and hundreds of cheaper, generic manufacturers—are designed to offer homeowners peace of mind. They allow a parent to check on a sleeping baby from the office, or a homeowner to see who is at the door from miles away. Example: Telegram motion-alert bot (high level)

However, the convenience of "plug-and-play" often comes at the cost of security. To make setup easy for the average consumer, many devices ship with default usernames and passwords (often as simple as "admin/admin" or "12345"). When users fail to change these, their cameras become open doors.

Telegram groups dedicated to these vulnerabilities are the digital equivalent of a peeping tom’s convention, but on a global scale. In these channels, thousands of members do not trade stolen credit card numbers or hacked accounts. They trade lives.

A typical feed in one of these groups is a collage of the mundane and the intimate. A grainy, green-tinted night vision shot of a couple sleeping. A high-definition view of a receptionist’s desk in a dental clinic in Brazil. A chaotic feed of children playing in a nursery in Ohio.

The posts are rarely captioned with context. They are simply labeled with a location—a country, sometimes a city—and occasionally a mocking note: "Nice kitchen," or "She doesn't know."

if name == "main": import sys

if len(sys.argv) > 1 and sys.argv[1] == "setup":
    create_config_template()
    create_requirements()
    print("\n📋 Setup complete! Follow these steps:")
    print("1. Edit config.json with your camera details")
    print("2. Install requirements: pip install -r requirements.txt")
    print("3. Run: python ipcam_monitor.py")
else:
    # Run the monitor
    monitor = IPCameraTelegramMonitor()
    asyncio.run(monitor.run())

You can find them via:

📷 *Welcome to the IPCam Enthusiasts Group* 📷
🔒 Rules:

📌 Useful resources:

👥 Share:

👉 Invite link: [INSERT YOUR GROUP LINK]
🚨 Admin: @username